fix(models): reduce repeated Nous recommendation traffic

This commit is contained in:
Robin Fernandes
2026-09-20 09:21:52 +10:00
committed by Teknium
parent 22beb95427
commit 5fa01f2fe0
2 changed files with 155 additions and 16 deletions

View File

@@ -10,6 +10,7 @@ from __future__ import annotations
import contextvars
import copy
import gzip
import json
import logging
import os
@@ -85,7 +86,10 @@ def _get_json(
call time so monkeypatching ``_urlopen_model_catalog_request`` still applies). Raises on failure."""
req = urllib.request.Request(url, headers=headers or {})
with (opener or _urlopen_model_catalog_request)(req, timeout=timeout, **open_kwargs) as resp:
return json.loads(resp.read().decode())
body = resp.read()
if req.get_header("Accept-encoding") == "gzip" and resp.headers.get("Content-Encoding", "").lower() == "gzip":
body = gzip.decompress(body)
return json.loads(body.decode())
def _read_json_cache(path: Path, *, errors=Exception) -> Optional[dict]:
@@ -322,8 +326,8 @@ def check_nous_free_tier(*, force_fresh: bool = False, cached_only: bool = False
NOUS_RECOMMENDED_MODELS_PATH = "/api/nous/recommended-models"
_NOUS_RECOMMENDED_CACHE_TTL: int = 600 # seconds (10 minutes)
# (result_dict, timestamp) keyed by portal_base_url so staging vs prod don't collide.
_nous_recommended_cache: dict[str, tuple[dict[str, Any], float]] = {}
# (result_dict, monotonic timestamp), scoped to the profile and portal.
_nous_recommended_cache: dict[tuple[str, str], tuple[dict[str, Any], float]] = {}
def _nous_recommended_disk_path() -> "Path":
@@ -331,13 +335,18 @@ def _nous_recommended_disk_path() -> "Path":
return get_hermes_home() / "cache" / "nous_recommended_cache.json"
def _read_nous_recommended_disk(base: str) -> dict[str, Any] | None:
"""Last-known-good payload for ``base`` from the per-base disk map
``{"<base>": {"data": {...}, "ts": <epoch>}}`` (staging and prod don't collide), or None."""
blob = _read_json_cache(_nous_recommended_disk_path(), errors=(OSError, json.JSONDecodeError))
def _read_nous_recommended_disk(base: str) -> tuple[dict[str, Any], float] | None:
"""Return the last good payload and its age for the freshness check."""
blob = _read_json_cache(_nous_recommended_disk_path(), errors=(OSError, json.JSONDecodeError, UnicodeDecodeError))
entry = (blob or {}).get(base)
data = entry.get("data") if isinstance(entry, dict) else None
return data if isinstance(data, dict) and data else None
if not isinstance(data, dict) or not data:
return None
try:
age = time.time() - float(entry.get("ts", 0))
except (TypeError, ValueError, OverflowError):
age = float("inf")
return data, age
def _write_nous_recommended_disk(base: str, data: dict[str, Any]) -> None:
@@ -347,7 +356,7 @@ def _write_nous_recommended_disk(base: str, data: dict[str, Any]) -> None:
return
path = _nous_recommended_disk_path()
try:
blob = _read_json_cache(path, errors=(OSError, json.JSONDecodeError)) or {}
blob = _read_json_cache(path, errors=(OSError, json.JSONDecodeError, UnicodeDecodeError)) or {}
blob[base] = {"data": data, "ts": time.time()}
_write_json_cache(path, blob, indent=2)
except OSError as exc:
@@ -359,18 +368,25 @@ def fetch_nous_recommended_models(
) -> dict[str, Any]:
"""Fetch the Portal's public ``/api/nous/recommended-models`` payload (no auth).
Cached per portal URL for ``_NOUS_RECOMMENDED_CACHE_TTL`` seconds in process (``force_refresh``
bypasses); a successful fetch is also persisted as last-known-good on disk, which serves a live
failure so a transient Portal hiccup doesn't drop the recommendations.
Reuse successful results for ``_NOUS_RECOMMENDED_CACHE_TTL`` seconds, including across
process restarts. ``force_refresh`` bypasses both caches. Stale disk data remains a fallback
on live failure; reading it never renews its freshness.
"""
base = (portal_base_url or "https://portal.nousresearch.com").rstrip("/")
now = time.monotonic()
cached = _nous_recommended_cache.get(base)
cache_key = (_pricing_profile_key(), base)
cached = _nous_recommended_cache.get(cache_key)
if not force_refresh and cached is not None and now - cached[1] < _NOUS_RECOMMENDED_CACHE_TTL:
return cached[0]
disk = _read_nous_recommended_disk(base)
if not force_refresh and disk is not None and 0 <= disk[1] < _NOUS_RECOMMENDED_CACHE_TTL:
data, age = disk
_nous_recommended_cache[cache_key] = (data, now - age)
return data
try:
data = _get_json(
f"{base}{NOUS_RECOMMENDED_MODELS_PATH}", timeout=timeout, headers={"Accept": "application/json"}
f"{base}{NOUS_RECOMMENDED_MODELS_PATH}", timeout=timeout,
headers={"Accept": "application/json", "Accept-Encoding": "gzip"}
)
if not isinstance(data, dict):
data = {}
@@ -379,8 +395,8 @@ def fetch_nous_recommended_models(
if data:
_write_nous_recommended_disk(base, data)
else:
data = _read_nous_recommended_disk(base) or data
_nous_recommended_cache[base] = (data, now)
data = disk[0] if disk is not None else data
_nous_recommended_cache[cache_key] = (data, now)
return data

View File

@@ -0,0 +1,123 @@
"""Recommendation requests reuse disk entries and negotiate gzip across real processes."""
import gzip
import json
import os
from pathlib import Path
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import pytest
@pytest.fixture
def portal():
requests = []
state = {"version": 1, "status": 200, "compress": True}
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
requests.append(self.headers.get("Accept-Encoding"))
body = json.dumps({"paidRecommendedModels": [{"modelName": f"model-{state['version']}"}]}).encode()
compressed = state["compress"] and requests[-1] == "gzip"
if compressed:
body = gzip.compress(body)
self.send_response(state["status"])
self.send_header("Content-Type", "application/json")
if compressed:
self.send_header("Content-Encoding", "gzip")
self.end_headers()
self.wfile.write(body)
def log_message(self, *_args):
pass
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield f"http://127.0.0.1:{server.server_port}", requests, state
finally:
server.shutdown()
server.server_close()
thread.join()
def fetch_in_new_process(home, base, *, force=False):
result = subprocess.run(
[sys.executable, "-c", "from hermes_cli.models import fetch_nous_recommended_models; "
"import json, sys; print(json.dumps(fetch_nous_recommended_models(sys.argv[1], force_refresh=sys.argv[2] == 'True')))",
base, str(force)],
cwd=Path(__file__).resolve().parents[2],
env={**os.environ, "HERMES_HOME": str(home)},
capture_output=True, text=True, check=True, timeout=30,
)
return json.loads(result.stdout)
def test_process_restarts_reuse_fresh_disk_but_force_and_expiry_fetch(tmp_path, portal):
base, requests, state = portal
home = tmp_path / "client"
first = fetch_in_new_process(home, base)
assert fetch_in_new_process(home, base) == first
assert requests == ["gzip"]
# A server may ignore Accept-Encoding; the refresh must also accept plain JSON.
state.update(version=2, compress=False)
second = fetch_in_new_process(home, base, force=True)
assert second and second != first
assert len(requests) == 2
path = home / "cache" / "nous_recommended_cache.json"
disk = json.loads(path.read_text(encoding="utf-8"))
disk[base]["ts"] = time.time() - 3600
path.write_text(json.dumps(disk), encoding="utf-8")
state["version"] = 3
assert fetch_in_new_process(home, base) != second
assert len(requests) == 3
# A failed refresh retains the last good payload without making it fresh on disk.
disk = json.loads(path.read_text(encoding="utf-8"))
disk[base]["ts"] = time.time() - 3600
path.write_text(json.dumps(disk), encoding="utf-8")
state["status"] = 503
assert fetch_in_new_process(home, base) == disk[base]["data"]
assert json.loads(path.read_text(encoding="utf-8"))[base]["ts"] == disk[base]["ts"]
# Bad local cache bytes/metadata must never stop a healthy API refresh.
state["status"] = 200
path.write_bytes(b"\xff")
assert fetch_in_new_process(home, base) == disk[base]["data"]
for timestamp in [None, True, "invalid", float("nan"), time.time() + 3600, 10 ** 400]:
disk[base]["ts"] = timestamp
path.write_text(json.dumps(disk), encoding="utf-8")
before = len(requests)
assert fetch_in_new_process(home, base) == disk[base]["data"]
assert len(requests) == before + 1
def test_disk_cache_is_scoped_to_home_and_portal(tmp_path, portal):
base, requests, state = portal
a, b = tmp_path / "a", tmp_path / "b"
first = fetch_in_new_process(a, base)
state["version"] = 2
assert fetch_in_new_process(b, base) != first
assert fetch_in_new_process(a, base) == first
assert len(requests) == 2
assert fetch_in_new_process(a, base + "/staging") != first
assert len(requests) == 3
# The same process can switch profiles under the multiplex gateway too.
from hermes_cli.models import fetch_nous_recommended_models, _nous_recommended_cache
from hermes_constants import set_hermes_home_override, reset_hermes_home_override
_nous_recommended_cache.clear()
for home, expected in [(a, first), (b, {"paidRecommendedModels": [{"modelName": "model-2"}]}), (a, first)]:
token = set_hermes_home_override(home)
try:
assert fetch_nous_recommended_models(base) == expected
finally:
reset_hermes_home_override(token)
assert len(requests) == 3
_nous_recommended_cache.clear()