perf: optimize gateway cold start from ~6.9s to ~460ms (#3918)

Channel lazy load: discover_enabled() only imports enabled channel
modules instead of all 18 modules with heavy SDKs (telegram, discord,
slack, etc). discover_all() now delegates to discover_enabled().

Lazy OpenAI client: defer AsyncOpenAI() + httpx construction to
_ensure_client() with asyncio.Lock double-checked locking. openai
and httpx imports moved from module-level into _ensure_client().

Minor: lazy Nanobot/RunResult and CronService exports via __getattr__.

Benchmark: 6910ms → 460ms (-93.3%)
This commit is contained in:
chengyongru
2026-05-20 12:02:23 +08:00
committed by Xubin Ren
parent 1391aa3d57
commit af9f8d54b8
9 changed files with 173 additions and 74 deletions
+19 -7
View File
@@ -70,28 +70,40 @@ class ChannelManager:
def _init_channels(self) -> None:
"""Initialize channels discovered via pkgutil scan + entry_points plugins."""
from nanobot.channels.registry import discover_all
from nanobot.channels.registry import discover_channel_names, discover_enabled
transcription_provider = self.config.channels.transcription_provider
transcription_key = self._resolve_transcription_key(transcription_provider)
transcription_base = self._resolve_transcription_base(transcription_provider)
transcription_language = self.config.channels.transcription_language
for name, cls in discover_all().items():
# Collect enabled module names first, then only import those.
# Channel configs live in ChannelsConfig's extra fields (via
# extra="allow"), so we enumerate candidates from pkgutil scan
# (cheap, no imports) and any plugin keys in __pydantic_extra__.
names = discover_channel_names()
candidate_names = set(names)
extra = getattr(self.config.channels, "__pydantic_extra__", None) or {}
candidate_names.update(extra.keys())
enabled_names: set[str] = set()
for name in candidate_names:
section = getattr(self.config.channels, name, None)
if section is None:
continue
enabled = (
if (
section.get("enabled", False)
if isinstance(section, dict)
else getattr(section, "enabled", False)
)
if not enabled:
):
enabled_names.add(name)
for name, cls in discover_enabled(enabled_names, _names=names).items():
section = getattr(self.config.channels, name, None)
if section is None:
continue
try:
kwargs: dict[str, Any] = {}
# Only the WebSocket channel currently hosts the embedded webui
# surface; other channels stay oblivious to these knobs.
if cls.name == "websocket":
if self._session_manager is not None:
kwargs["session_manager"] = self._session_manager
+36 -14
View File
@@ -1,5 +1,4 @@
"""Auto-discovery for built-in channel modules and external plugins."""
from __future__ import annotations
import importlib
@@ -51,21 +50,44 @@ def discover_plugins() -> dict[str, type[BaseChannel]]:
return plugins
def discover_enabled(
enabled_names: set[str],
*,
_names: list[str] | None = None,
_include_all_external: bool = False,
) -> dict[str, type[BaseChannel]]:
"""Return channels whose module names are in *enabled_names*.
Uses cheap ``pkgutil.iter_modules`` to list names, then imports only
those that match — skipping the heavy third-party SDK imports of
unneeded channels.
"""
names = _names if _names is not None else discover_channel_names()
result: dict[str, type[BaseChannel]] = {}
for modname in names:
if modname not in enabled_names:
continue
try:
result[modname] = load_channel_class(modname)
except ImportError as e:
logger.debug("Skipping built-in channel '{}': {}", modname, e)
external = discover_plugins()
shadowed = set(external) & set(result)
if shadowed:
logger.warning("Plugin(s) shadowed by built-in channels (ignored): {}", shadowed)
if _include_all_external:
result.update({k: v for k, v in external.items() if k not in shadowed})
else:
result.update({k: v for k, v in external.items() if k not in shadowed and k in enabled_names})
return result
def discover_all() -> dict[str, type[BaseChannel]]:
"""Return all channels: built-in (pkgutil) merged with external (entry_points).
Built-in channels take priority — an external plugin cannot shadow a built-in name.
"""
builtin: dict[str, type[BaseChannel]] = {}
for modname in discover_channel_names():
try:
builtin[modname] = load_channel_class(modname)
except ImportError as e:
logger.debug("Skipping built-in channel '{}': {}", modname, e)
external = discover_plugins()
shadowed = set(external) & set(builtin)
if shadowed:
logger.warning("Plugin(s) shadowed by built-in channels (ignored): {}", shadowed)
return {**external, **builtin}
names = discover_channel_names()
return discover_enabled(set(names), _names=names, _include_all_external=True)