feat: llm_router shim — tri-level config priority (local > user > env-var)
This commit is contained in:
parent
b79d13b4f2
commit
f62a9d9901
2 changed files with 163 additions and 4 deletions
|
|
@ -1,19 +1,46 @@
|
|||
"""
|
||||
LLM abstraction layer with priority fallback chain.
|
||||
Reads config/llm.yaml. Tries backends in order; falls back on any error.
|
||||
Config lookup order:
|
||||
1. <repo>/config/llm.yaml — per-install local config
|
||||
2. ~/.config/circuitforge/llm.yaml — user-level config (circuitforge-core default)
|
||||
3. env-var auto-config (ANTHROPIC_API_KEY, OPENAI_API_KEY, OLLAMA_HOST, …)
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
from circuitforge_core.llm import LLMRouter as _CoreLLMRouter
|
||||
|
||||
# Kept for backwards-compatibility — external callers that import CONFIG_PATH
|
||||
# from this module continue to work.
|
||||
CONFIG_PATH = Path(__file__).parent.parent / "config" / "llm.yaml"
|
||||
|
||||
|
||||
class LLMRouter(_CoreLLMRouter):
|
||||
"""Peregrine-specific LLMRouter — defaults to Peregrine's config/llm.yaml."""
|
||||
"""Peregrine-specific LLMRouter — tri-level config path priority.
|
||||
|
||||
def __init__(self, config_path: Path = CONFIG_PATH):
|
||||
super().__init__(config_path)
|
||||
When ``config_path`` is supplied (e.g. in tests) it is passed straight
|
||||
through to the core. When omitted, the lookup order is:
|
||||
1. <repo>/config/llm.yaml (per-install local config)
|
||||
2. ~/.config/circuitforge/llm.yaml (user-level, circuitforge-core default)
|
||||
3. env-var auto-config (ANTHROPIC_API_KEY, OPENAI_API_KEY, OLLAMA_HOST …)
|
||||
"""
|
||||
|
||||
def __init__(self, config_path: Path | None = None) -> None:
|
||||
if config_path is not None:
|
||||
# Explicit path supplied — use it directly (e.g. tests, CLI override).
|
||||
super().__init__(config_path)
|
||||
return
|
||||
|
||||
local = Path(__file__).parent.parent / "config" / "llm.yaml"
|
||||
user_level = Path.home() / ".config" / "circuitforge" / "llm.yaml"
|
||||
if local.exists():
|
||||
super().__init__(local)
|
||||
elif user_level.exists():
|
||||
super().__init__(user_level)
|
||||
else:
|
||||
# No yaml found — let circuitforge-core's env-var auto-config run.
|
||||
# The core default CONFIG_PATH (~/.config/circuitforge/llm.yaml)
|
||||
# won't exist either, so _auto_config_from_env() will be triggered.
|
||||
super().__init__()
|
||||
|
||||
|
||||
# Module-level singleton for convenience
|
||||
|
|
|
|||
132
tests/test_llm_router_shim.py
Normal file
132
tests/test_llm_router_shim.py
Normal file
|
|
@ -0,0 +1,132 @@
|
|||
"""Tests for Peregrine's LLMRouter shim — priority fallback logic."""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock, call
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
|
||||
def _import_fresh():
|
||||
"""Import scripts.llm_router fresh (bypass module cache)."""
|
||||
import importlib
|
||||
import scripts.llm_router as mod
|
||||
importlib.reload(mod)
|
||||
return mod
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 1: local config/llm.yaml takes priority when it exists
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_uses_local_yaml_when_present():
|
||||
"""When config/llm.yaml exists locally, super().__init__ is called with that path."""
|
||||
import scripts.llm_router as shim_mod
|
||||
from circuitforge_core.llm import LLMRouter as _CoreLLMRouter
|
||||
|
||||
local_path = Path(shim_mod.__file__).parent.parent / "config" / "llm.yaml"
|
||||
user_path = Path.home() / ".config" / "circuitforge" / "llm.yaml"
|
||||
|
||||
def fake_exists(self):
|
||||
return self == local_path # only the local path "exists"
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_core_init(self, config_path=None):
|
||||
captured["config_path"] = config_path
|
||||
self.config = {}
|
||||
|
||||
with patch.object(Path, "exists", fake_exists), \
|
||||
patch.object(_CoreLLMRouter, "__init__", fake_core_init):
|
||||
import importlib
|
||||
import scripts.llm_router as mod
|
||||
importlib.reload(mod)
|
||||
mod.LLMRouter()
|
||||
|
||||
assert captured.get("config_path") == local_path, (
|
||||
f"Expected super().__init__ to be called with local path {local_path}, "
|
||||
f"got {captured.get('config_path')}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 2: falls through to env-var auto-config when neither yaml exists
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_falls_through_to_env_when_no_yamls():
|
||||
"""When no yaml files exist, super().__init__ is called with no args (env-var path)."""
|
||||
import scripts.llm_router as shim_mod
|
||||
from circuitforge_core.llm import LLMRouter as _CoreLLMRouter
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_exists(self):
|
||||
return False # no yaml files exist anywhere
|
||||
|
||||
def fake_core_init(self, config_path=None):
|
||||
# Record whether a path was passed
|
||||
captured["config_path"] = config_path
|
||||
captured["called"] = True
|
||||
self.config = {}
|
||||
|
||||
with patch.object(Path, "exists", fake_exists), \
|
||||
patch.object(_CoreLLMRouter, "__init__", fake_core_init):
|
||||
import importlib
|
||||
import scripts.llm_router as mod
|
||||
importlib.reload(mod)
|
||||
mod.LLMRouter()
|
||||
|
||||
assert captured.get("called"), "super().__init__ was never called"
|
||||
# When called with no args, config_path defaults to None in our mock,
|
||||
# meaning the shim correctly fell through to env-var auto-config
|
||||
assert captured.get("config_path") is None, (
|
||||
f"Expected super().__init__ to be called with no explicit path (None), "
|
||||
f"got {captured.get('config_path')}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3: module-level complete() singleton is only instantiated once
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_complete_singleton_is_reused():
|
||||
"""complete() reuses the same LLMRouter instance across multiple calls."""
|
||||
import importlib
|
||||
import scripts.llm_router as mod
|
||||
importlib.reload(mod)
|
||||
|
||||
# Reset singleton
|
||||
mod._router = None
|
||||
|
||||
instantiation_count = [0]
|
||||
original_init = mod.LLMRouter.__init__
|
||||
|
||||
mock_router = MagicMock()
|
||||
mock_router.complete.return_value = "OK"
|
||||
|
||||
original_class = mod.LLMRouter
|
||||
|
||||
class CountingRouter(original_class):
|
||||
def __init__(self):
|
||||
instantiation_count[0] += 1
|
||||
# Bypass real __init__ to avoid needing config files
|
||||
self.config = {}
|
||||
|
||||
def complete(self, prompt, system=None):
|
||||
return "OK"
|
||||
|
||||
# Patch the class in the module
|
||||
mod.LLMRouter = CountingRouter
|
||||
mod._router = None
|
||||
|
||||
result1 = mod.complete("first call")
|
||||
result2 = mod.complete("second call")
|
||||
|
||||
assert result1 == "OK"
|
||||
assert result2 == "OK"
|
||||
assert instantiation_count[0] == 1, (
|
||||
f"Expected LLMRouter to be instantiated exactly once, "
|
||||
f"got {instantiation_count[0]} instantiation(s)"
|
||||
)
|
||||
|
||||
# Restore
|
||||
mod.LLMRouter = original_class
|
||||
Loading…
Reference in a new issue