跳到主要内容

可插拔机制:注入、回调、技能

这章讲让 OpenHands「同一套代码、不同后端」的三个底层机制。它们也是顶层第 7 步(事件回流 / 自动化)与「换后端」承诺的实现基础。

1. Injector:在配置里选实现​

要解决的小问题: 沙箱可以是 Docker / 远程 / 进程,事件可以存文件系统 / SQL / 云……业务代码(如会话启动)不该 import 具体实现,否则换后端就得改代码。

思路: 每个子系统拆成「抽象服务 + 一个 Injector」。Injector 是个产出服务实例的工厂,以 async 生成器形式提供,天然适配 FastAPI 的 Depends 与「请求级生命周期」:

# openhands/app_server/services/injector.py(节选)
class Injector(Generic[T], ABC):
    @abstractmethod
    async def inject(self, state, request=None) -> AsyncGenerator[T, None]:
        """产出一个服务实例;state 可在嵌套注入间共享。"""
        yield None

    async def depends(self, request: Request) -> AsyncGenerator[T, None]:
        async for result in self.inject(request.state, request):
            yield result   # 直接喂给 FastAPI Depends

关键在于:SandboxService 等抽象类各自带一个 ...Injector(DiscriminatedUnionMixin, Injector[...])(如 SandboxServiceInjector,sandbox/sandbox_service.py:247)。DiscriminatedUnionMixin 是 SDK 提供的判别联合基类——它让「选哪个实现」变成一个可以从环境变量 / 配置 JSON 反序列化出来的值。config.py 顶部把所有这些 Injector 类型 import 进来(openhands/app_server/config.py:46-52 等),启动时根据配置实例化出对应实现。

一句话:抽象服务定义「能做什么」,Injector + 判别联合决定「用哪个实现来做」,配置说了算。

2. EventCallbackProcessor:自动化的底座​

要解决的小问题: agent 跑出来的事件(收到消息、动作完成…)要能触发副作用——自动改会话标题、发 Slack、拆 GitHub issue。这些副作用必须可配置、可持久化、可按事件类型路由。

思路: 用一个抽象处理器 EventCallbackProcessor,同样基于 DiscriminatedUnionMixin——于是一条「回调」可以连同它的处理器一起存进数据库,之后从 JSON 还原出正确的处理器子类:

# event_callback/event_callback_models.py:40(EventCallbackProcessor)
class EventCallbackProcessor(DiscriminatedUnionMixin, ABC):
    event_kind: ClassVar[EventKind] = 'MessageEvent'   # 关心哪种事件

    @abstractmethod
    async def __call__(self, conversation_id, callback, event) -> EventCallbackResult | None:
        """处理一个事件。"""

注意 EventKind 本身是从 SDK 的 Event 全部具体子类动态生成的字面量联合(event_callback_models.py:30,用 get_known_concrete_subclasses(Event))——所以「事件类型」永远跟 SDK 对齐,不会写死过期。

注册一个回调 = 存一条 EventCallback{conversation_id, processor, event_kind}(:90)。仓库自带的处理器示例:LoggingCallbackProcessor(打日志,:57)、set_title_callback_processor.py(自动生成会话标题)。

分发逻辑 在 SQL 实现里:按「会话 + 事件类型 + ACTIVE」查出所有回调,并发执行,失败的记成 ERROR 结果而不炸整批:

# event_callback/sql_event_callback_service.py:208(execute_callbacks)
query = (select(StoredEventCallback)
         .where(StoredEventCallback.status == EventCallbackStatus.ACTIVE)
         .where(StoredEventCallback.event_kind == event.kind)      # 按事件类型路由
         .where(StoredEventCallback.conversation_id == conversation_id))
...
await asyncio.gather(*[self.execute_callback(conversation_id, cb, event) for cb in callbacks])

单个回调出错被 try/except 兜住、记 ERROR、继续跑其它(execute_callback,:233-250)。这就是 README 里「自动化、定时任务、webhook 响应」对外能力的内部支点:外部 webhook 进 webhook_router → 转成事件 → execute_callbacks 路由到处理器。

3. 多源技能加载与合并​

要解决的小问题: 给 agent 注入「技能 / 微 agent(microagent)」——可复用的提示片段 / 工具说明。它们散落在多个地方:公共库、用户目录、组织仓库、项目仓库、沙箱自身。要统一加载并去重。

思路:控制中心不自己解析技能,而是调沙箱里 agent-server 的 /api/skills 端点,一次 API 调用合并所有来源。 它只负责「告诉 agent-server 去哪些地方找」。来源清单见 app_conversation_service_base.py:104-112(load_and_merge_all_skills 的 docstring):

来源位置
公共技能OpenHands/skills GitHub 仓库
用户技能~/.openhands/skills/
组织技能{org}/.openhands 仓库
项目技能repo 的 .agents/skills/、.openhands/microagents/、旧版 .openhands/skills/
沙箱技能沙箱暴露 URL

控制中心这边做的是配置准备:build_org_configs 解析组织仓库路径(GitHub 系有 .openhands 和 .agents 两个独立仓库,GitLab / Azure DevOps 只有一个 openhands-config,见 skill_loader.py:198-216 的 _candidate_repo_paths),build_sandbox_config 收集沙箱 URL,然后一把 POST 给 agent-server:

# skill_loader.py:393(load_skills_from_agent_server)
# 单次 API 调用,让 agent-server 加载并合并所有来源的技能

合并去重 由控制中心在拿到结果后兜底:_merge_skills 按名字去重,后面的列表覆盖前面的(项目技能可覆盖公共技能):

# app_conversation_service_base.py:191-208(_merge_skills)
skills_by_name: dict[str, Skill] = {}
for skill_list in skill_lists:      # 列表顺序 = 优先级,后者覆盖前者
    for skill in skill_list:
        skills_by_name[skill.name] = skill
return list(skills_by_name.values())

去重只发生在 _merge_skills;disabled_skills 的过滤是另一个方法 _load_skills_and_update_agent 干的——它先调 load_and_merge_all_skills 拿到合并结果,再按名字把被禁用的技能剔除:

# app_conversation_service_base.py:239-242(_load_skills_and_update_agent)
if disabled_skills:
    disabled_set = set(disabled_skills)
    all_skills = [s for s in all_skills if s.name not in disabled_set]

合并(并过滤)后的技能塞进 agent 的 AgentContext.skills(_create_agent_with_skills,:163),随会话请求发给 agent-server。

4. 三者的共同暗线​

这三个机制共享同一个设计模式,值得单独点出:

  • 注入 和 回调 都建立在 SDK 的 DiscriminatedUnionMixin 上——「实现选择」被编码成可序列化的数据(配置 / DB 行),而非硬编码的 import。
  • 技能 和 回调 都遵循「控制中心只准备配置、把活外包」:技能解析外包给 agent-server 的 /api/skills,agent 执行外包给 agent-server 的 /api/conversations。
  • 一以贯之的是顶层那条控制面 / 数据面分离(见 01-top-level.md):控制中心保持「薄」,所有重活在沙箱或 SDK 里。

5. 代码地图​

主题文件符号
依赖注入基类openhands/app_server/services/injector.pyInjector、depends
服务+注入器配对(示例)openhands/app_server/sandbox/sandbox_service.pySandboxServiceInjector
事件回调处理器抽象openhands/app_server/event_callback/event_callback_models.pyEventCallbackProcessor、EventCallback
事件类型动态联合同上EventKind(get_known_concrete_subclasses)
回调分发openhands/app_server/event_callback/sql_event_callback_service.pyexecute_callbacks、execute_callback
自动改标题处理器openhands/app_server/event_callback/set_title_callback_processor.py(处理器实现)
技能多源加载openhands/app_server/app_conversation/skill_loader.pyload_skills_from_agent_server、build_org_configs、_candidate_repo_paths
技能合并去重openhands/app_server/app_conversation/app_conversation_service_base.py_merge_skills、_load_skills_and_update_agent、_create_agent_with_skills、load_and_merge_all_skills