refactor(team_runner): replace static adapter imports with dynamic importlib loading

Concrete adapter classes (AnthropicAdapter, GitHubAdapter, etc.) are no
longer imported at the top of team_runner.py. Instead, each registry maps
short names to 'module.path:ClassName' strings resolved lazily via
importlib.import_module at instantiation time.

This means:
- Adding a new adapter requires only an entry in the registry string dict
  (or a full dotted path directly in team.yaml) — no changes to TeamRunner.
- Third-party / custom adapters work out of the box: set e.g.
  adapters.llm: mypackage.llm.openai:OpenAIAdapter in team.yaml.
- The runner no longer hard-wires knowledge of which concrete classes exist.

Addresses tandrewng review comment on PR #1.
This commit is contained in:
2026-03-16 00:30:28 -04:00
parent bd96a83069
commit 71316b3090

View File

@@ -25,15 +25,12 @@ from core.blackboard import Blackboard
from core.escalation import EscalationHandler
from core.task_brief import TaskBrief
import importlib
from adapters.base.llm import LLMAdapter
from adapters.base.notify import NotifyAdapter
from adapters.base.runtime import RuntimeAdapter
from adapters.base.vcs import VCSAdapter
from adapters.llm.anthropic import AnthropicAdapter
from adapters.notify.openclaw import OpenClawNotifyAdapter
from adapters.runtime.claude_code import ClaudeCodeRuntimeAdapter
from adapters.runtime.openclaw import OpenClawRuntimeAdapter
from adapters.vcs.github import GitHubAdapter
logger = logging.getLogger(__name__)
@@ -59,22 +56,60 @@ _TIER_CAPABILITIES: dict[int, str] = {
5: "fast-cheap",
}
# Adapter registry: config key → concrete class.
_LLM_ADAPTERS: dict[str, type[LLMAdapter]] = {
"anthropic": AnthropicAdapter,
# ---------------------------------------------------------------------------
# Adapter registries
#
# Values are "module.path:ClassName" strings resolved lazily via importlib.
# To add a new adapter, append an entry here — no changes to TeamRunner needed.
# team.yaml may also supply a full "module.path:ClassName" value directly,
# enabling third-party adapters without touching this file.
# ---------------------------------------------------------------------------
_LLM_ADAPTERS: dict[str, str] = {
"anthropic": "adapters.llm.anthropic:AnthropicAdapter",
}
_VCS_ADAPTERS: dict[str, type[VCSAdapter]] = {
"github": GitHubAdapter,
_VCS_ADAPTERS: dict[str, str] = {
"github": "adapters.vcs.github:GitHubAdapter",
}
_NOTIFY_ADAPTERS: dict[str, type[NotifyAdapter]] = {
"openclaw": OpenClawNotifyAdapter,
_NOTIFY_ADAPTERS: dict[str, str] = {
"openclaw": "adapters.notify.openclaw:OpenClawNotifyAdapter",
}
_RUNTIME_ADAPTERS: dict[str, type[RuntimeAdapter]] = {
"openclaw": OpenClawRuntimeAdapter,
"claude_code": ClaudeCodeRuntimeAdapter,
_RUNTIME_ADAPTERS: dict[str, str] = {
"openclaw": "adapters.runtime.openclaw:OpenClawRuntimeAdapter",
"claude_code": "adapters.runtime.claude_code:ClaudeCodeRuntimeAdapter",
}
def _load_adapter_class(key: str, registry: dict[str, str], label: str) -> type:
"""
Resolve a short name or dotted "module:ClassName" path to an adapter class.
Resolution order:
1. If *key* is in *registry*, use the mapped dotted path.
2. Otherwise, treat *key* itself as a dotted path (custom / third-party).
"""
dotted = registry.get(key, key)
if ":" not in dotted:
raise ValueError(
f"Unknown {label} adapter {key!r}. "
f"Built-in choices: {list(registry)}. "
f"Or supply a full 'module.path:ClassName' value in team.yaml."
)
module_path, class_name = dotted.rsplit(":", 1)
try:
module = importlib.import_module(module_path)
except ModuleNotFoundError as exc:
raise ImportError(
f"Cannot import {label} adapter module {module_path!r}: {exc}"
) from exc
try:
return getattr(module, class_name)
except AttributeError:
raise ImportError(
f"Module {module_path!r} has no class {class_name!r}"
)
# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------
@@ -163,28 +198,24 @@ class TeamRunner:
return fh.read()
def _build_llm(self, key: str) -> LLMAdapter:
cls = _LLM_ADAPTERS.get(key)
if cls is None:
raise ValueError(
f"Unknown LLM adapter {key!r}. Known: {list(_LLM_ADAPTERS)}"
)
cls = _load_adapter_class(key, _LLM_ADAPTERS, "LLM")
return cls(self._config)
def _build_optional(
self,
registry: dict,
registry: dict[str, str],
key: Optional[str],
label: str,
) -> Optional[object]:
"""Build an optional adapter, returning None on any init error."""
if not key:
return None
cls = registry.get(key)
if cls is None:
logger.warning("Unknown %s adapter %r — skipping.", label, key)
return None
try:
cls = _load_adapter_class(key, registry, label)
return cls(self._config)
except (ImportError, ValueError) as exc:
logger.warning("Unknown %s adapter %r — skipping. (%s)", label, key, exc)
return None
except Exception as exc:
logger.warning(
"%s adapter %r could not be initialised (%s) — skipping.",
@@ -195,11 +226,7 @@ class TeamRunner:
return None
def _build_runtime(self, key: str) -> RuntimeAdapter:
cls = _RUNTIME_ADAPTERS.get(key)
if cls is None:
raise ValueError(
f"Unknown runtime adapter {key!r}. Known: {list(_RUNTIME_ADAPTERS)}"
)
cls = _load_adapter_class(key, _RUNTIME_ADAPTERS, "runtime")
return cls(self._config)
# ------------------------------------------------------------------