Newer
Older
navi-1 / navi / core / registry.py
@Eugene Sukhodolskiy Eugene Sukhodolskiy on 9 May 8 KB Add profile editing to admin panel
"""Registries for tools, profiles, and LLM backends."""

from navi.config import settings
from navi.exceptions import ProfileNotFound, ToolNotFound
from navi.llm.base import LLMBackend
from navi.llm.ollama import OllamaBackend
from navi.llm.fallback import FallbackOllamaBackend, load_servers_from_file
from navi.profiles import ALL_PROFILES
from navi.profiles.base import AgentProfile
from navi.tools import (
    CodeExecTool,
    DeleteToolTool,
    FilesystemTool,
    HttpRequestTool,
    ImageViewTool,
    ListProfilesTool,
    MemoryTool,
    ReflectTool,
    ScadLintTool,
    SpawnAgentTool,
    SshExecTool,
    ScratchpadTool,
    SwitchProfileTool,
    TerminalTool,
    TestToolTool,
    TodoTool,
    Tool,
    WebSearchTool,
    WebViewTool,
)
from navi.tools.list_tools import ListToolsTool
from navi.tools.reload_tools import ReloadToolsTool
from navi.tools.tool_manual import ToolManualTool
from navi.tools.write_tool import WriteToolTool
from navi.tools.share_file import ShareFileTool
from navi.tools.content_publish import ContentPublishTool
from navi.tools.model_3d import Model3DTool
from navi.tools.render_3d import Render3DTool
from navi.tools.loader import LoadResult, load_tools_from_dir
from navi.tools.logging_middleware import LoggingMiddleware
from navi.context_providers._loader import ContextProviderRegistry


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}
        self._builtin_names: set[str] = set()
        self._middlewares: list = []

    def register(self, tool: Tool, builtin: bool = False) -> None:
        self._tools[tool.name] = tool
        if builtin:
            self._builtin_names.add(tool.name)

    def add_middleware(self, middleware) -> None:
        """Add a ToolMiddleware instance."""
        self._middlewares.append(middleware)

    def get(self, name: str) -> Tool:
        if name not in self._tools:
            raise ToolNotFound(name)
        return self._tools[name]

    def resolve(self, names: list[str]) -> list[Tool]:
        return [self.get(n) for n in names]

    def all(self) -> list[Tool]:
        return list(self._tools.values())

    def reload_user_tools(self, tools_dir: str) -> LoadResult:
        """Remove all user tools and reload from disk. Safe: errors are isolated."""
        # Drop previously loaded user tools
        for name in list(self._tools):
            if name not in self._builtin_names:
                del self._tools[name]

        result = load_tools_from_dir(tools_dir)
        for tool in result.loaded:
            self._tools[tool.name] = tool
        return result


class ProfileRegistry:
    def __init__(self) -> None:
        self._profiles: dict[str, AgentProfile] = {}

    def register(self, profile: AgentProfile) -> None:
        self._profiles[profile.id] = profile

    def get(self, profile_id: str) -> AgentProfile:
        if profile_id not in self._profiles:
            raise ProfileNotFound(profile_id)
        return self._profiles[profile_id]

    def all(self) -> list[AgentProfile]:
        return list(self._profiles.values())

    def update(self, profile: AgentProfile) -> None:
        """Replace an existing profile in-memory."""
        if profile.id not in self._profiles:
            raise ProfileNotFound(profile.id)
        self._profiles[profile.id] = profile


class BackendRegistry:
    def __init__(self) -> None:
        self._backends: dict[str, LLMBackend] = {}

    def register(self, key: str, backend: LLMBackend) -> None:
        self._backends[key] = backend

    def get(self, key: str) -> LLMBackend:
        backend = self._backends.get(key)
        if backend is None:
            raise KeyError(f"LLM backend '{key}' not registered")
        return backend

    def all_keys(self) -> list[str]:
        return list(self._backends.keys())


def _discover_backends() -> list[tuple[str, LLMBackend]]:
    """Auto-discover LLM backends from navi/llm/ modules."""
    discovered: list[tuple[str, LLMBackend]] = []
    from navi.llm.ollama import OllamaBackend
    from navi.llm.fallback import FallbackOllamaBackend
    from navi.llm.openai_backend import OpenAIBackend
    ollama_http_timeout = max(
        settings.ollama_request_timeout,
        settings.llm_complete_timeout,
        settings.llm_stream_first_chunk_timeout,
    )

    # Ollama backend (primary)
    if settings.ollama_backends_file:
        servers = load_servers_from_file(settings.ollama_backends_file)
        discovered.append(("ollama", FallbackOllamaBackend(servers)))
    else:
        discovered.append(("ollama", OllamaBackend(
            model=settings.ollama_default_model,
            host=settings.ollama_host,
            api_key=settings.ollama_api_key,
            timeout=ollama_http_timeout,
        )))

    # OpenAI backend (if configured)
    if settings.openai_api_key:
        discovered.append(("openai", OpenAIBackend(
            model=settings.openai_model,
            api_key=settings.openai_api_key,
            base_url=settings.openai_base_url,
        )))

    return discovered


def build_default_registries(
    memory_store=None,
    session_store=None,
) -> tuple[ToolRegistry, ProfileRegistry, BackendRegistry, ContextProviderRegistry]:
    """Build and populate registries with all built-in components."""
    from navi.core.ai_helper import AIHelper

    backends = BackendRegistry()
    backend_instances = _discover_backends()
    if not backend_instances:
        raise RuntimeError("No LLM backends discovered. Check OLLAMA_HOST or OPENAI_API_KEY.")

    for key, backend in backend_instances:
        backends.register(key, backend)

    # Use primary backend for AIHelper
    primary_backend = backend_instances[0][1]
    ai_helper = AIHelper(
        backend=primary_backend,
        default_model=settings.ollama_default_model,
    )

    tools = ToolRegistry()
    reload_tool = ReloadToolsTool(registry=tools)
    write_tool = WriteToolTool(registry=tools)
    delete_tool = DeleteToolTool(registry=tools)
    list_tool = ListToolsTool(registry=tools)
    manual_tool = ToolManualTool(registry=tools)
    memory_tool = MemoryTool(memory_store) if memory_store else None
    builtins = [WebSearchTool(), FilesystemTool(ai_helper=ai_helper), HttpRequestTool(), WebViewTool(),
                CodeExecTool(), TerminalTool(), SshExecTool(), ImageViewTool(), ScadLintTool(),
                ShareFileTool(), ContentPublishTool(), TestToolTool(),
                Model3DTool(), Render3DTool(),
                TodoTool(), ScratchpadTool(), ReflectTool(ai_helper=ai_helper),
                reload_tool, write_tool, delete_tool, list_tool, manual_tool]
    if memory_tool:
        builtins.append(memory_tool)
    for builtin in builtins:
        tools.register(builtin, builtin=True)

    # User-defined tools loaded from tools_dir
    result = load_tools_from_dir(settings.tools_dir)
    for user_tool in result.loaded:
        tools.register(user_tool)

    # Register built-in middleware
    tools.add_middleware(LoggingMiddleware())

    profiles = ProfileRegistry()
    for p in ALL_PROFILES:
        profiles.register(p)

    # Tools that need session_store + profile_registry registered after both are built.
    spawn_tool = SpawnAgentTool(
        profile_registry=profiles,
        tool_registry=tools,
        backend_registry=None,   # patched below after backends are built
        session_store=session_store,
        memory_store=memory_store,
    )
    tools.register(spawn_tool, builtin=True)

    switch_tool = SwitchProfileTool(
        session_store=session_store,
        profile_registry=profiles,
    )
    tools.register(switch_tool, builtin=True)

    list_profiles_tool = ListProfilesTool(profile_registry=profiles)
    tools.register(list_profiles_tool, builtin=True)

    # Patch backend registry into spawn_tool now that it's available
    spawn_tool._backend_registry = backends

    # Context providers
    cp_registry = ContextProviderRegistry()
    from pathlib import Path as _Path
    cp_registry.load_from_dir(str(_Path(__file__).parent.parent / "context_providers"), builtin=True)
    cp_registry.load_from_dir(settings.context_providers_dir, builtin=False)
    reload_tool._cp_registry = cp_registry

    return tools, profiles, backends, cp_registry