AI Agent框架探秘：拆解 OpenHands（10）--- Runtime

数察啜 · 昨天 21:30

AI Agent框架探秘：拆解 OpenHands（10）--- Runtime

目录

AI Agent框架探秘：拆解 OpenHands（10）--- Runtime
- 0x00 摘要
- 0x01 工作机制
  - 1.1 Environment 和 Sandbox
  - 1.2 安全执行
  - 1.3 解决方案
  - 1.4 核心功能
  - 1.5 工作机制
- 0x02 核心逻辑
  - 2.1. base.py
  - 2.2 ActionExecutionClient
  - 2.3 运行时类型
    - 2.3.1 Docker运行时环境
    - 2.3.2 本地运行时环境
    - 2.3.3 远程运行时环境
    - 2.3.4 单例模式
  - 2.4 工作流程
  - 2.5. Runtime与其他组件关系
    - 2.5.1 EventStream
    - 2.5.2 AgentController
    - 2.5.3 Session
    - 2.5.4 插件系统
    - 2.5.5 文件系统和存储
    - 2.5.6 git仓库
    - 2.5.7 Python & MCP
- 0x03 代码
  - 3.1 定义&初始化
  - 3.2 关键代码
    - 3.2.1 环境初始化
    - 3.2.2 事件处理
    - 3.2.3 微代理支持
- 0xFF 参考

0x00 摘要

对于Agent ，Google 白皮书给出一个简洁而实用的定义：Agent = 模型 + 工具 + 编排层 + 部署运行时，这里和目前大部分的 AI Agent 的定义（LLM + Tool + Memory）多了一层部署运行时。因此可见Runtime的重要性。
在 OpenHands 里，真正让“AI 想法”落地就是Runtime。它像一座可移动的实验室：四面墙把主机世界隔开，却给 Agent 留下齐全的操作台（文件、终端、网络）。Agent 只需递上一张写满指令的纸条，Runtime 便接管全部脏活：启动进程、捕捉回显、隔离风险、回收资源，再把结果封装成“观察报告”送回前台。由于所有动作都在统一沙箱里完成，外部系统既不用担心权限越界，也不必为环境差异头疼。
简言之，Runtime 是决策与执行之间的安全转换器，它是Agent与外部环境交互的桥梁，也是整个系统稳定运转的压舱石，保障了整个系统的高效协同运作。
因为本系列借鉴的文章过多，可能在参考文献中有遗漏的文章，如果有，还请大家指出。
0x01 工作机制

在当今数字化时代，代码的安全执行已成为至关重要的议题。OpenHands 项目为应对这一挑战，精心设计了一套基于 Docker 容器的沙盒系统，旨在为代码执行提供一个既安全又隔离的环境。这一系统不仅保障了代码的安全性，还确保了执行的一致性和可复现性，同时实现了资源的有效控制和不同项目之间的隔离。
1.1 Environment 和 Sandbox

Environment 指的是 Agent 可操作的容器，相当于给了 Agent 一台可自行操作的计算机，Agent 可以在其中端到端地完成任务，这个赛道包括 Sandbox、Browser Infra、Agent 操作系统等不同的细分领域。
其中，Sandbox 是一种安全机制，为执行中的程序提供隔离环境，即为 Agent 提供了一个可以隔离运行的虚拟机环境，开发者可以在这个环境中实现 Agent 的开发、部署、运行。但随着 Agent 的不断发展，传统的虚拟机并不能很好地满足 Agent 需求，原因在于：

Agent 对虚拟机的性能提出了更高的要求，比如需要更高的隔离性、更快的启动速度、更强的稳定性；
Agent 的虚拟机还要求具备一定的 AI 性能，例如需要具备代码解释器（code interpretor）的功能，或需要整合开发者常用的 AI 架构，如 Vercel AI SDK。

1.2 安全执行

代码安全执行的五大理由

安全性：在执行不受信任的代码时，必须确保这些代码不会对主机系统造成损害。沙盒环境通过严格的访问控制，防止恶意代码访问或修改主机系统的资源，从而保护主机免受潜在威胁。
一致性：沙盒环境确保代码在不同机器和配置下的执行结果具有一致性。这种一致性消除了“在我的机器上可以运行”的常见问题，使得代码在任何环境下都能稳定运行。
资源控制：通过沙盒化，可以精确控制资源的分配和使用。这不仅防止了失控的进程对主机系统造成影响，还确保了资源的合理分配，提升了系统的整体性能。
隔离性：不同的项目或用户可以在各自的隔离环境中工作，互不干扰。这种隔离性不仅保护了主机系统，还确保了不同项目之间的独立性，避免了资源竞争和潜在的冲突。
可复现性：沙盒环境的一致性和可控性使得复现错误和问题变得更为容易。这在调试和问题解决过程中尤为重要，因为一致的环境可以确保问题的重现和解决。

1.3 解决方案

OpenHands 通过基于 Docker 容器的沙盒环境，为代码执行构建了一道坚固的“安全防线”。每个项目在启动时，系统会自动创建一个独立的 Docker 容器作为其专属沙盒。这个沙盒拥有隔离的文件系统、网络环境和资源配额，确保代码只能访问容器内部的文件，网络请求被限制在预设的安全域内，CPU 和内存的使用也受到严格管控。
这种隔离性带来了三重保障：

避免项目间干扰：某一项目的代码错误（如无限循环导致的内存溢出）只会影响其所在的沙盒，不会波及其他项目或主机系统。
防范恶意代码风险：即使 AI 生成的代码中包含潜在的危险操作（如删除系统文件），也会被沙盒环境拦截，无法对主机造成实质损害。
简化环境一致性管理：沙盒的基础镜像可以预先配置好特定版本的编程语言、依赖库和工具链，确保代码在开发、测试、生产环境中的执行结果一致，避免了“在我电脑上能运行”的问题。

尽管沙盒环境提供了高度的安全性和隔离性，但它并非一个孤立的“孤岛”。OpenHands 框架提供了精细的“端口映射”和“目录挂载”机制，允许开发者将沙盒内的特定端口暴露给主机（便于调试），或将主机的指定目录挂载到沙盒中（用于输入输出文件传递）。这种机制在确保安全性的同时，也提供了足够的灵活性，使得开发和调试过程更加高效。
通过这种设计，OpenHands 不仅确保了代码的安全执行，还提升了开发效率和环境管理的便利性。这种平衡是现代软件开发中不可或缺的一部分，为开发者提供了一个既安全又高效的开发环境。
1.4 核心功能

Runtime的核心功能可以概括为四个主要方面：

工作环境的构建与管理：Runtime负责创建和管理代理的工作区域，无论是隔离性更强的容器环境还是便捷的本地环境，都能根据需求提供定制化的工作空间，确保代理在执行任务时不会受到外部因素的干扰。
动作的执行：代理发出的指令，如文件编辑、命令执行等，都由Runtime解析并精确执行，它充当了决策与实践之间的桥梁。
环境变量的维护：Runtime负责维护任务执行所需的环境变量，为任务执行提供必要的配置支持。
环境的生命周期管理：Runtime全程管理环境的生命周期，从初始化到断开连接，形成完整的闭环。同时，通过EventStream实时输出执行日志和观察结果，为控制器、记忆系统、MCP等组件提供关键的状态反馈，确保整个系统的协同运作。

1.5 工作机制

OpenHands 运行时系统采用基于 Docker 容器实现的客户端 - 服务器架构，其工作机制概述如下：

用户输入：用户提供一个自定义的基础 Docker 镜像。
镜像构建：OpenHands 以用户提供的镜像为基础，构建一个新的 Docker 镜像（即 “OH 运行时镜像”）。该新镜像包含 OpenHands 专属代码，核心为 “运行时客户端”。
容器启动：当 OpenHands 启动时，会使用 OH 运行时镜像启动一个 Docker 容器。
动作执行服务器初始化：动作执行服务器在容器内部初始化一个 ActionExecutor（动作执行器），配置必要组件（如 Bash Shell）并加载指定的插件。
通信过程：OpenHands 后端通过 RESTful API 与动作执行服务器通信，发送动作指令并接收执行反馈数据。
动作执行：运行时客户端接收来自后端的动作指令，在沙箱环境中执行这些指令，并将执行反馈数据回传。
反馈数据返回：动作执行服务器将执行结果以反馈数据（Observation）的形式发送回 OpenHands 后端。

客户端的核心作用：

作为 OpenHands 后端与沙箱环境之间的中间媒介，实现双向数据传递。
在容器内安全执行各类动作指令（包括 Shell 命令、文件操作、Python 代码等）。
管理沙箱环境的状态，涵盖当前工作目录和已加载插件等信息。
对反馈数据进行格式化处理后返回给后端，为结果处理提供统一的接口规范。

工作机制参见下图。

0x02 核心逻辑

Runtime是在用户交互期间为用户的智能体应用程序提供动力的底层引擎。它是一个系统，接收用户定义的智能体、工具和回调，并协调它们对用户输入的执行，管理信息流、状态变化以及与外部服务（如 LLM 或存储）的交互。可以将运行时视为你的智能体应用程序的"引擎"。用户定义部件（智能体、工具），而运行时处理它们如何连接并一起运行以满足用户请求。
Runtime支持多种执行环境，包括Docker容器、本地环境等，使Agent能够安全地执行代码和命令。其派生类有：DockerRuntime、RemoteRuntime、LocalRuntime、KubernetesRuntime、CLIRuntime。
核心功能

命令执行：提供Bash shell访问能力。
浏览器交互：支持网页浏览和交互操作。
文件系统操作：文件读写，编辑等操作。
git 操作管理：仓库克隆、分支管理、变更跟踪。
环境变量管理：运行时环境变量配置。
插件系统管理：支持VSCode、Jupyter等插件集成。

2.1. base.py

base.py 文件定义了Runtime类，它作为代理与外部环境交互的主要接口。它处理各种操作，包括：

沙盒执行
浏览器交互
文件系统操作
环境变量管理
插件管理

Runtime类的关键特性：

使用配置和事件流进行初始化
使用ainit方法异步初始化环境变量
针对不同类型的操作执行方法（运行、读取、写入、浏览等）
文件操作的抽象方法（由子类实现）

2.2 ActionExecutionClient

action_execution_client.py 文件包含ActionExecutionClient类，它实现了运行时接口。它是一个抽象实现，意味着仍需要通过具体实现来扩展才能使用。
此客户端通过HTTP调用与action_execution_server交互以实际执行运行时操作。

class ActionExecutionClient(Runtime):
"""Base class for runtimes that interact with the action execution server.
This class contains shared logic between DockerRuntime and RemoteRuntime
for interacting with the HTTP server defined in action_execution_server.py.
"""
def __init__(
self,
config: OpenHandsConfig,
event_stream: EventStream,
llm_registry: LLMRegistry,
sid: str = 'default',
plugins: list[PluginRequirement] | None = None,
env_vars: dict[str, str] | None = None,
status_callback: Any | None = None,
attach_to_existing: bool = False,
headless_mode: bool = True,
user_id: str | None = None,
git_provider_tokens: PROVIDER_TOKEN_TYPE | None = None,
):
self.session = HttpSession()
self.action_semaphore = threading.Semaphore(1) # Ensure one action at a time
self._runtime_closed: bool = False
self._vscode_token: str | None = None # initial dummy value
self._last_updated_mcp_stdio_servers: list[MCPStdioServerConfig] = []
super().__init__(
config,
event_stream,
llm_registry,
sid,
plugins,
env_vars,
status_callback,
attach_to_existing,
headless_mode,
user_id,
git_provider_tokens,
)

复制代码

其余的所有Runtime均继承 ActionExecutionClient

class KubernetesRuntime(ActionExecutionClient):
class LocalRuntime(ActionExecutionClient):
class RemoteRuntime(ActionExecutionClient):
class DockerRuntime(ActionExecutionClient):

复制代码

2.3 运行时类型

2.3.1 Docker运行时环境

为使用Docker容器设计的Docker运行时环境，OpenHands做如下配置：

为每个会话创建和管理Docker容器
在容器内执行动作
支持直接文件系统访问和本地资源管理
适合开发、测试和需要完全控制执行环境的场景

Docker运行时环的关键特性为：

实时日志和调试能力
直接访问本地文件系统
由于本地资源执行速度更快
容器隔离以提高安全性

2.3.2 本地运行时环境

本地运行时环境设计用于本地机器上的直接执行。目前仅支持以本地用户身份运行：

直接在主机上运行action_execution_server
无Docker容器开销
直接访问本地系统资源
适合Docker不可用或不需要开发和测试

关键特性：

需要的设置最少
直接访问本地资源
无容器开销
最快执行速度

重要：此运行时不提供隔离，因为它直接在主机上运行。所有动作以运行OpenHands的用户的相同权限执行。对于需要适当隔离的安全执行，请改用Docker运行时。
2.3.3 远程运行时环境

远程运行时环境设计用于远程环境中的执行：

连接到运行ActionExecutor的远程服务器
通过向远程客户端发送请求执行动作
支持分布式执行和基于云的部署
适合生产环境、可扩展性和本地资源限制是问题的场景

关键特性：

可扩展性和资源灵活性
减少本地资源使用
支持基于云的部署
通过隔离提高安全性的潜力

目前，这主要用于并行评估，例如这个SWE-Bench示例。
2.3.4 单例模式

Runtime的子类不是单例模式。从函数_create_runtime 可以看出，每次运行都会创建一个新的实例。每个Runtime都有自己的状态，如sid（会话ID）。每个实例也可以调用close()来清理资源。

async def _create_runtime(
self,
runtime_name: str,
config: OpenHandsConfig,
agent: Agent,
git_provider_tokens: PROVIDER_TOKEN_TYPE | None = None,
custom_secrets: CUSTOM_SECRETS_TYPE | None = None,
selected_repository: str | None = None,
selected_branch: str | None = None,
) -> bool:
"""Creates a runtime instance
Parameters:
- runtime_name: The name of the runtime associated with the session
- config:
- agent:
Return True on successfully connected, False if could not connect.
Raises if already created, possibly in other situations.
"""
self.runtime = runtime_cls(
config=config,
event_stream=self.event_stream,
llm_registry=self.llm_registry,
sid=self.sid,
plugins=agent.sandbox_plugins,
status_callback=self._status_callback,
headless_mode=False,
attach_to_existing=False,
env_vars=env_vars,
git_provider_tokens=git_provider_tokens,
)
return True

复制代码

2.4 工作流程

Runtime作为EventStreamSubscriber.RUNTIME订阅者，处理来自事件流的Action并生成Observation。Runtime的工作流程如下：

初始化：
- Runtime使用配置和事件流进行初始化。
- 设置环境变量。
- 加载并初始化插件。
动作处理：
- Runtime通过事件流接收动作。
- 验证并路由到适当的执行方法。
动作执行：
- 执行不同类型的动作：
  - 使用run方法执行bash命令
  - 使用run_ipython方法执行IPython单元
  - 使用read和write方法执行文件操作
  - 使用browse和browse_interactive方法浏览网页
观察生成：
- 动作执行后，生成相应的观察结果。
- 观察结果被添加到事件流中。
插件集成：
- 插件如Jupyter和AgentSkills被初始化并集成到运行时。
沙盒环境：
- ActionExecutor在Docker容器内设置沙盒环境。
- 初始化用户环境和bash shell。
- 从OpenHands后端接收的动作在此沙盒环境中执行。
浏览器交互：
- 使用BrowserEnv类处理网络浏览动作。

2.5. Runtime与其他组件关系

Runtime与其他组件的主要关系如下：

运行时与openhands.events模块中定义的事件系统紧密交互。
依赖于openhands.core.config中的配置类。
日志通过openhands.core.logger处理。

我们详细看看。
2.5.1 EventStream

Runtime通过与EventStream与其他模块进行事件驱动的交互。

class Runtime(FileEditRuntimeMixin):
# Runtime订阅事件流
event_stream.subscribe(
EventStreamSubscriber.RUNTIME, self.on_event, self.sid
)
# Runtime处理传入的事件
def on_event(self, event: Event) -> None:
if isinstance(event, Action):
asyncio.get_event_loop().run_until_complete(self._handle_action(event)) # Runtime返回观察结果给事件流
self.event_stream.add_event(observation, source)

复制代码

2.5.2 AgentController

AgentController通过EventStream向Runtime发送操作命令，Runtime执行后将结果返回。

AgentController 发送Action → EventStream
Runtime 接到 Action 后执行
Rutime 发送 Observation → EventStream
AgentController 接收到 Observation 并继续执行决策流程

2.5.3 Session

WebSession 和 AgentSession 负责管理 Runtime 的生命周期

# AgentSession
runtime_cls = get_runtime_cls(runtime_name)
# 创建Runtime实例
self.runtime = runtime_cls(
config=config,
event_stream=self.event_stream,
llm_registry=self.llm_registry,
sid=self.sid,
plugins=agent.sandbox_plugins,
status_callback=self._status_callback,
headless_mode=False,
attach_to_existing=False,
git_provider_tokens=overrided_tokens,
env_vars=env_vars,
user_id=self.user_id,
)
# 连接到Runtime
await self.runtime.connect()
# 关闭Runtime
EXECUTOR.submit(self.runtime.close)

复制代码

2.5.4 插件系统

Runtime管理和执行各种插件功能。

# 初始化加载插件
self.plugins = (
copy.deepcopy(plugins) if plugins is not None and len(plugins) > 0 else []
)
# add VSCode plugin if not in headless mode
if not headless_mode:
self.plugins.append(VSCodeRequirement())
# 执行插件相关操作
if any(isinstance(plugin, JupyterRequirement) for plugin in self.plugins):
code = 'import os\n'
for key, value in env_vars.items():
# Note: json.dumps gives us nice escaping for free
code += f'os.environ["{key}"] = {json.dumps(value)}\n'
code += '\n'
self.run_ipython(IPythonRunCellAction(code))
......

复制代码

2.5.5 文件系统和存储

Runtime 提供文件操作能力。

def read(self, action: FileReadAction) -> Observation:
def write(self, action: FileWriteAction) -> Observation:

复制代码

2.5.6 git仓库

Runtime 提供git仓库操作能力。

async def clone_or_init_repo(
self,
git_provider_tokens: PROVIDER_TOKEN_TYPE | None,
selected_repository: str | None,
selected_branch: str | None,
) -> str:

复制代码

2.5.7 Python & MCP

Runtime 提供运行python代码和call_tool_mcp操作能力。

def run_ipython(self, action: IPythonRunCellAction) -> Observation:
async def call_tool_mcp(self, action: MCPAction) -> Observation:

复制代码

0x03 代码

3.1 定义&初始化

class Runtime(FileEditRuntimeMixin):
"""智能代理运行时环境的抽象基类。
这是OpenHands中的一个扩展点，允许应用程序自定义代理与外部环境的交互方式。
该运行时提供一个沙箱环境，包含以下功能：
- Bash shell访问
- 浏览器交互
- 文件系统操作
- Git操作
- 环境变量管理
应用程序可通过以下方式替换为自定义实现：
1. 创建一个继承自Runtime的类
2. 实现所有必需的方法
3. 在配置中设置运行时名称或使用get_runtime_cls()方法
该类通过get_runtime_cls()中的get_impl()方法实例化。
内置实现包括：
- DockerRuntime：基于Docker的容器化环境
- RemoteRuntime：远程执行环境
- LocalRuntime：用于开发的本地执行环境
- KubernetesRuntime：基于Kubernetes的执行环境
- CLIRuntime：命令行界面运行时
参数：
sid：唯一标识当前用户会话的会话ID
"""
sid: str # 会话ID，唯一标识用户会话
config: OpenHandsConfig # OpenHands配置对象
initial_env_vars: dict[str, str] # 初始环境变量字典
attach_to_existing: bool # 是否连接到现有运行时环境
status_callback: Callable[[str, RuntimeStatus, str], None] | None # 状态回调函数，接收会话ID、运行时状态和消息
runtime_status: RuntimeStatus | None # 当前运行时状态
_runtime_initialized: bool = False # 运行时初始化状态标记
security_analyzer: 'SecurityAnalyzer | None' = None # 安全分析器实例，用于检测潜在风险
def __init__(
self,
config: OpenHandsConfig,
event_stream: EventStream,
llm_registry: LLMRegistry,
sid: str = 'default',
plugins: list[PluginRequirement] | None = None,
env_vars: dict[str, str] | None = None,
status_callback: Callable[[str, RuntimeStatus, str], None] | None = None,
attach_to_existing: bool = False,
headless_mode: bool = False,
user_id: str | None = None,
git_provider_tokens: PROVIDER_TOKEN_TYPE | None = None,
):
# 初始化Git处理器，绑定shell执行和文件创建的回调方法
self.git_handler = GitHandler(
execute_shell_fn=self._execute_shell_fn_git_handler, # Git操作所需的shell执行函数
create_file_fn=self._create_file_fn_git_handler, # Git操作所需的文件创建函数
)
# 初始化会话ID
self.sid = sid
# 绑定事件流（组件间通信的核心）
self.event_stream = event_stream
# 若事件流存在，订阅运行时相关事件
if event_stream:
event_stream.subscribe(
EventStreamSubscriber.RUNTIME, # 订阅者类型（运行时）
self.on_event, # 事件处理回调函数
self.sid # 订阅者ID（当前会话ID）
)
# 初始化插件列表（深拷贝传入的插件，为空时设为默认空列表）
self.plugins = (
copy.deepcopy(plugins) if plugins is not None and len(plugins) > 0 else []
)
# 非无头模式下添加VSCode插件
if not headless_mode:
self.plugins.append(VSCodeRequirement())
# 绑定状态回调函数
self.status_callback = status_callback
# 记录是否连接到现有运行时
self.attach_to_existing = attach_to_existing
# 深拷贝配置对象（避免外部修改影响内部状态）
self.config = copy.deepcopy(config)
# 注册程序退出时的关闭回调（确保资源正确释放）
atexit.register(self.close)
# 初始化默认环境变量（基于沙箱配置）
self.initial_env_vars = _default_env_vars(config.sandbox)
# 合并用户传入的环境变量（覆盖默认值）
if env_vars is not None:
self.initial_env_vars.update(env_vars)
# 初始化Provider处理器（管理Git等服务的访问令牌）
self.provider_handler = ProviderHandler(
provider_tokens=git_provider_tokens
or cast(PROVIDER_TOKEN_TYPE, MappingProxyType({})), # 令牌为空时使用空映射
external_auth_id=user_id, # 外部认证ID（关联用户）
external_token_manager=True, # 启用外部令牌管理
)
# 同步调用异步方法获取Provider相关环境变量
raw_env_vars: dict[str, str] = call_async_from_sync(
self.provider_handler.get_env_vars, # 获取环境变量的异步方法
GENERAL_TIMEOUT, # 超时时间
True, # 允许重试
None, # 重试间隔（使用默认）
False # 不抛出异常
)
# 合并Provider返回的环境变量
self.initial_env_vars.update(raw_env_vars)
# 检查是否启用VSCode插件（通过判断插件列表中是否包含VSCodeRequirement实例）
self._vscode_enabled = any(
isinstance(plugin, VSCodeRequirement) for plugin in self.plugins
)
# 初始化文件编辑混合类（提供文件编辑相关功能）
FileEditRuntimeMixin.__init__(
self,
enable_llm_editor=config.get_agent_config().enable_llm_editor, # 是否启用LLM辅助编辑
llm_registry=llm_registry, # LLM注册中心（用于获取语言模型实例）
)
# 记录用户ID
self.user_id = user_id
# 记录Git服务提供商令牌
self.git_provider_tokens = git_provider_tokens
# 初始化运行时状态（默认为None）
self.runtime_status = None
# 初始化安全分析器（若配置启用）
self.security_analyzer = None
if self.config.security.security_analyzer:
# 根据配置获取安全分析器类（默认使用SecurityAnalyzer）
analyzer_cls = options.SecurityAnalyzers.get(
self.config.security.security_analyzer, SecurityAnalyzer
)
# 实例化安全分析器
self.security_analyzer = analyzer_cls()
# 为安全分析器绑定事件流（用于发布安全相关事件）
self.security_analyzer.set_event_stream(self.event_stream)

复制代码

3.2 关键代码

3.2.1 环境初始化

setup_initial_env提供了环境初始化能力

def setup_initial_env(self) -> None:
if self.attach_to_existing:
return
logger.debug(f'Adding env vars: {self.initial_env_vars.keys()}')
self.add_env_vars(self.initial_env_vars)
if self.config.sandbox.runtime_startup_env_vars:
self.add_env_vars(self.config.sandbox.runtime_startup_env_vars)
# Configure git settings
self._setup_git_config()

复制代码

3.2.2 事件处理

on_event函数接受来自事件流的 Action，_handle_action函数执行对应的操作，生成Observation。

def on_event(self, event: Event) -> None:
if isinstance(event, Action):
asyncio.get_event_loop().run_until_complete(self._handle_action(event))

复制代码

3.2.3 微代理支持

get_microagents_from_selected_repo 从仓库加载微代理配置
get_microagents_from_org_or_user

比如get_microagents_from_org_or_user是 OpenHands 系统中组织 / 用户级微智能体加载的核心逻辑，负责从代码仓库的组织或用户级配置仓库中加载微智能体（轻量级智能体组件）。主要功能包括：

仓库路径解析：从目标仓库路径中提取组织 / 用户名，确定配置仓库位置。
平台适配：区分 GitLab 和其他平台（如 GitHub），使用不同的配置仓库名称（GitLab 用 openhands-config，其他用 .openhands）。
仓库克隆：通过带认证的 URL 克隆配置仓库，采用浅克隆（--depth 1）提高效率。
微智能体加载：从克隆仓库的 microagents 目录加载微智能体，并在加载完成后清理临时文件。
异常处理：针对认证失败、克隆错误等场景进行日志记录，确保流程稳健性。

流程如下

代码如下：

def get_microagents_from_org_or_user(
self, selected_repository: str
) -> List[BaseMicroagent]:
"""从组织或用户级仓库加载微智能体。
例如：若目标仓库为 github.com/acme-co/api，会检查 github.com/acme-co/.openhands 是否存在。
若存在，会克隆该仓库并从 ./microagents/ 文件夹加载微智能体。
对于 GitLab 仓库，会使用 openhands-config 而非 .openhands，因为 GitLab 不支持
以非字母数字字符开头的仓库名称。
参数：
selected_repository: 仓库路径（例如："github.com/acme-co/api"）
返回：
从组织/用户级仓库加载的微智能体列表
"""
loaded_microagents: List[BaseMicroagent] = []
# 拆分仓库路径为多个部分（按 '/' 分割）
repo_parts = selected_repository.split('/')
# 校验路径格式：至少需要包含域名、组织/用户名、仓库名三部分（如 github.com/org/repo）
if len(repo_parts) < 2:
return loaded_microagents
# 提取组织/用户名（路径中倒数第二部分）
org_name = repo_parts[-2]
# 判断是否为 GitLab 仓库
is_gitlab = self._is_gitlab_repository(selected_repository)
# 确定组织级配置仓库名称：GitLab 用 openhands-config，其他用 .openhands
if is_gitlab:
org_openhands_repo = f'{org_name}/openhands-config'
else:
org_openhands_repo = f'{org_name}/.openhands'
# 尝试克隆组织级配置仓库
try:
# 创建组织仓库的临时目录（避免冲突）
org_repo_dir = self.workspace_root / f'org_openhands_{org_name}'
# 获取带认证的仓库 URL 并执行浅克隆（--depth 1 提高效率）
try:
# 同步调用异步方法获取认证 URL（带超时控制）
remote_url = call_async_from_sync(
self.provider_handler.get_authenticated_git_url,
GENERAL_TIMEOUT,
org_openhands_repo,
)
except AuthenticationError as e:
raise # 重新抛出认证异常，终止当前流程
except Exception as e:
raise # 重新抛出其他异常
# 构建克隆命令：禁用终端交互提示，浅克隆到临时目录
clone_cmd = (
f'GIT_TERMINAL_PROMPT=0 git clone --depth 1 {remote_url} {org_repo_dir}'
)
# 执行克隆命令
action = CmdRunAction(command=clone_cmd)
obs = self.run_action(action)
# 检查克隆结果：退出码为 0 表示成功
if isinstance(obs, CmdOutputObservation) and obs.exit_code == 0:
# 从组织仓库的 microagents 目录加载微智能体
org_microagents_dir = org_repo_dir / 'microagents'
loaded_microagents = self._load_microagents_from_directory(
org_microagents_dir, 'org-level'
)
# 清理临时目录：加载完成后删除克隆的仓库
action = CmdRunAction(f'rm -rf {org_repo_dir}')
self.run_action(action)
else:
# 克隆失败：提取错误信息和退出码
clone_error_msg = (
obs.content
if isinstance(obs, CmdOutputObservation)
else 'Unknown error'
)
exit_code = (
obs.exit_code if isinstance(obs, CmdOutputObservation) else 'N/A'
)
return loaded_microagents

复制代码

0xFF 参考

https://docs.all-hands.dev/openhands/usage/architecture/runtime
Agent Infra 图谱：哪些组件值得为 Agent 重做一遍？

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

AI Agent框架探秘：拆解 OpenHands（10）--- Runtime

相关帖子

浏览过的版块

签约作者

AI Agent框架探秘：拆解 OpenHands（10）--- Runtime

相关帖子

相关推荐

浏览过的版块

签约作者