laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-08-09 02:54:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(upstream): resolve generic provider pricing instead of fabricating it

Browse Source 
A generic (OpenAI-compatible) upstream whose /models response omits pricing
was silently defaulting to $0.001/M tokens and a 4096 context window. For a
provider like DeepSeek that reports no price, this undercharged real usage by
~280x — a direct money leak — while presenting a plausible-looking price.

Resolve each model through trust-ordered sources instead: the provider's
native schema (Venice's model_spec) first, then litellm's bundled cost map
(curated list prices), then the OpenRouter feed. Capture the richer metadata
those sources carry (cache rates, modalities, max output tokens, context)
rather than only price and context. When no source knows the model, import it
disabled with a warning rather than invent a number, so an operator can price
it before it serves traffic.

Context has no trustworthy source of last resort, but it is not a billing
input, so a model priced without a reported context window falls back to an
id-based estimate. The whole source-incomplete fallback (price chain +
context estimate) lives in one pricing_resolver module so it can later be
hoisted into the base provider unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jeroen Ubbink
2026-07-03 14:42:33 +02:00
co-authored by Claude Opus 4.8
parent ffde661d93
commit 2a27fb239e
3 changed files with 534 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/upstream/generic.py
+82 -44
View File
@@ -5,6 +5,11 @@ from typing import TYPE_CHECKING
import httpx
from .base import BaseUpstreamProvider
from .pricing_resolver import (
FallbackPricingResolver,
ResolvedPricing,
estimate_context_length,
)
if TYPE_CHECKING:
from ..core.db import UpstreamProviderRow
@@ -64,6 +69,35 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
"platform_url": cls.platform_url,
}
def _native_pricing(
self, model_id: str, model_spec: dict
) -> ResolvedPricing | None:
"""Read pricing/metadata from Venice's bespoke ``model_spec`` schema.
Returns ``None`` when the upstream reported no native price (the common
case for bare OpenAI-compatible ``/models`` responses), so the caller
falls through to the shared resolution chain instead of fabricating a
number.
"""
pricing_info = model_spec.get("pricing", {})
input_usd = pricing_info.get("input", {}).get("usd")
output_usd = pricing_info.get("output", {}).get("usd")
if input_usd is None or output_usd is None:
return None
capabilities = model_spec.get("capabilities", {})
input_modalities = ["text"]
if capabilities.get("supportsVision", False):
input_modalities.append("image")
return ResolvedPricing(
prompt=input_usd / 1_000_000,
completion=output_usd / 1_000_000,
context_length=model_spec.get("availableContextTokens"),
source="native",
input_modalities=input_modalities,
)
async def fetch_models(self) -> list[Model]:
"""Fetch models from upstream API using /models endpoint."""
from ..payment.models import Architecture, Model, Pricing, TopProvider
@@ -78,6 +112,7 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
response.raise_for_status()
data = response.json()
resolver = FallbackPricingResolver()
models_list = []
for model_data in data.get("data", []):
model_id = model_data.get("id", "")
@@ -89,41 +124,41 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
owned_by = model_data.get("owned_by", "unknown")
model_spec = model_data.get("model_spec", {})
context_length = 4096
if model_spec.get("availableContextTokens"):
context_length = model_spec["availableContextTokens"]
elif any(
pattern in model_id.lower() for pattern in ["32k", "32000"]
):
context_length = 32768
elif any(
pattern in model_id.lower() for pattern in ["16k", "16000"]
):
context_length = 16384
elif any(pattern in model_id.lower() for pattern in ["8k", "8000"]):
context_length = 8192
elif "gpt-4" in model_id.lower():
context_length = 8192
elif "claude" in model_id.lower():
context_length = 200000
resolved = self._native_pricing(model_id, model_spec)
if resolved is None:
resolved = await resolver.resolve(model_id)
pricing_info = model_spec.get("pricing", {})
input_pricing = pricing_info.get("input", {})
output_pricing = pricing_info.get("output", {})
if resolved is None:
# Fail closed: never invent a price. Import the model
# disabled with a warning so the operator can price it
# (the admin UI surfaces disabled remote models).
logger.warning(
f"No pricing source resolved for '{model_id}' from "
f"{self.upstream_name}; importing it disabled",
extra={"model_id": model_id, "base_url": self.base_url},
)
resolved = ResolvedPricing(
prompt=0.0,
completion=0.0,
context_length=None,
source="unresolved",
)
enabled = False
else:
enabled = True
prompt_price = input_pricing.get("usd", 0.001) / 1000000
completion_price = output_pricing.get("usd", 0.001) / 1000000
modality = (
"text->text"
if "image" in resolved.input_modalities
else "text"
)
capabilities = model_spec.get("capabilities", {})
input_modalities = ["text"]
output_modalities = ["text"]
if capabilities.get("supportsVision", False):
input_modalities.append("image")
modality = "text"
if capabilities.get("supportsVision", False):
modality = "text->text"
# A source can carry a price but no context (e.g. a litellm
# entry missing max_input_tokens); fall back to an id-based
# estimate so we never persist a zero-length window.
context_length = resolved.context_length or estimate_context_length(
model_id
)
spec_name = model_spec.get("name", model_name)
description = f"{spec_name}"
@@ -139,30 +174,33 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
context_length=context_length,
architecture=Architecture(
modality=modality,
input_modalities=input_modalities,
output_modalities=output_modalities,
tokenizer="unknown",
instruct_type=None,
input_modalities=resolved.input_modalities,
output_modalities=resolved.output_modalities,
tokenizer=resolved.tokenizer,
instruct_type=resolved.instruct_type,
),
pricing=Pricing(
prompt=prompt_price,
completion=completion_price,
prompt=resolved.prompt,
completion=resolved.completion,
request=0.0,
image=0.0,
web_search=0.0,
internal_reasoning=0.0,
max_prompt_cost=0.001,
max_completion_cost=0.001,
max_cost=0.001,
input_cache_read=resolved.input_cache_read,
input_cache_write=resolved.input_cache_write,
),
sats_pricing=None,
per_request_limits=None,
top_provider=TopProvider(
context_length=context_length,
max_completion_tokens=context_length // 2,
is_moderated=False,
max_completion_tokens=(
resolved.max_completion_tokens
if resolved.max_completion_tokens is not None
else context_length // 2
),
is_moderated=bool(resolved.is_moderated),
),
enabled=True,
enabled=enabled,
upstream_provider_id=None,
canonical_slug=None,
)
routstr/upstream/pricing_resolver.py
+177
View File
@@ -0,0 +1,177 @@
"""Shared price/metadata resolution chain for upstream model discovery.
Most OpenAI-compatible ``/models`` responses carry no pricing. Rather than let
a provider fabricate one, this module resolves a model through decreasingly
trustworthy sources — litellm's bundled cost map (curated list prices, mirrors
provider docs), then the OpenRouter feed (resale prices, broader coverage) —
and returns ``None`` when none of them know the model, so the caller can fail
closed instead of inventing a number.
Provider-native pricing (a gateway's own ``/models`` schema, e.g. Venice's
``model_spec``) is authoritative and handled by the provider before this chain
is consulted; only the shared fallback lives here so a later refactor can hoist
it into the base provider unchanged.
"""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class ResolvedPricing:
"""Per-token pricing plus whatever metadata the answering source carried.
Prices are USD per token. ``source`` records provenance
(``native``/``litellm``/``openrouter``/``unresolved``) so later work can
surface where each price came from.
"""
prompt: float
completion: float
context_length: int | None
source: str
max_completion_tokens: int | None = None
input_cache_read: float = 0.0
input_cache_write: float = 0.0
input_modalities: list[str] = field(default_factory=lambda: ["text"])
output_modalities: list[str] = field(default_factory=lambda: ["text"])
tokenizer: str = "unknown"
instruct_type: str | None = None
is_moderated: bool | None = None
def estimate_context_length(model_id: str) -> int:
"""Best-effort context window from a model id when no source reports one.
The last rung of the fallback chain, reached only for a model whose price
resolved but whose context did not (or that imported disabled). Context is
not a billing input, so a rough id-based guess is acceptable here where a
guessed *price* never would be.
"""
lowered = model_id.lower()
if any(pattern in lowered for pattern in ["32k", "32000"]):
return 32768
if any(pattern in lowered for pattern in ["16k", "16000"]):
return 16384
if any(pattern in lowered for pattern in ["8k", "8000"]):
return 8192
if "gpt-4" in lowered:
return 8192
if "claude" in lowered:
return 200000
return 4096
def _as_float(value: object) -> float | None:
"""OpenRouter reports prices as strings; coerce, ``None`` if unparseable."""
try:
return float(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return None
def _as_int(value: object) -> int | None:
"""Coerce an already-numeric token count to ``int``, else ``None``."""
return int(value) if isinstance(value, (int, float)) else None
def _from_litellm(model_id: str) -> ResolvedPricing | None:
# Lazy import so the resolver stays import-light and shares the exact
# lookup semantics used by cache-rate backfill.
from ..payment.models import litellm_cost_entry
info = litellm_cost_entry(model_id)
if info is None:
return None
prompt = info.get("input_cost_per_token")
completion = info.get("output_cost_per_token")
if not isinstance(prompt, (int, float)) or not isinstance(completion, (int, float)):
return None
input_modalities = ["text"]
if info.get("supports_vision"):
input_modalities.append("image")
return ResolvedPricing(
prompt=float(prompt),
completion=float(completion),
context_length=_as_int(info.get("max_input_tokens") or info.get("max_tokens")),
source="litellm",
max_completion_tokens=_as_int(info.get("max_output_tokens")),
input_cache_read=float(info.get("cache_read_input_token_cost") or 0.0),
input_cache_write=float(info.get("cache_creation_input_token_cost") or 0.0),
input_modalities=input_modalities,
)
def _match_openrouter(model_id: str, feed: list[dict]) -> dict | None:
"""Find ``model_id`` in the OpenRouter feed, exact id before bare tail.
Bare-tail matching (``deepseek-chat`` ↔ ``deepseek/deepseek-chat``) is a
looser, lower-trust match — OpenRouter fans a model out across resellers —
so an exact id match always wins first.
"""
bare = model_id.split("/", 1)[-1]
exact = next((m for m in feed if m.get("id") == model_id), None)
if exact is not None:
return exact
return next(
(m for m in feed if m.get("id", "").split("/", 1)[-1] == bare), None
)
def _from_openrouter(model_id: str, feed: list[dict]) -> ResolvedPricing | None:
entry = _match_openrouter(model_id, feed)
if entry is None:
return None
pricing = entry.get("pricing", {})
prompt = _as_float(pricing.get("prompt"))
completion = _as_float(pricing.get("completion"))
if prompt is None or completion is None:
return None
architecture = entry.get("architecture", {})
top_provider = entry.get("top_provider", {})
return ResolvedPricing(
prompt=prompt,
completion=completion,
context_length=_as_int(entry.get("context_length")),
source="openrouter",
max_completion_tokens=_as_int(top_provider.get("max_completion_tokens")),
input_cache_read=_as_float(pricing.get("input_cache_read")) or 0.0,
input_cache_write=_as_float(pricing.get("input_cache_write")) or 0.0,
input_modalities=architecture.get("input_modalities") or ["text"],
output_modalities=architecture.get("output_modalities") or ["text"],
tokenizer=architecture.get("tokenizer") or "unknown",
instruct_type=architecture.get("instruct_type"),
is_moderated=top_provider.get("is_moderated"),
)
class FallbackPricingResolver:
"""Resolves models via litellm → OpenRouter for one discovery pass.
The OpenRouter catalog is fetched at most once and only when a model
actually misses litellm, so a provider full of litellm-known models never
touches the network. Instantiate one per ``fetch_models`` call.
"""
def __init__(self) -> None:
self._openrouter_feed: list[dict] | None = None
async def resolve(self, model_id: str) -> ResolvedPricing | None:
"""Resolve ``model_id``; ``None`` if no source knows it."""
resolved = _from_litellm(model_id)
if resolved is not None:
return resolved
if self._openrouter_feed is None:
# Lazy import so tests can patch the feed at its source.
from ..payment.models import async_fetch_openrouter_models
self._openrouter_feed = await async_fetch_openrouter_models()
return _from_openrouter(model_id, self._openrouter_feed)
tests/unit/test_upstream_generic.py
+275
View File
@@ -0,0 +1,275 @@
"""Unit tests for ``GenericUpstreamProvider.fetch_models`` price/metadata resolution.
A generic upstream is any OpenAI-compatible API. Most (DeepSeek, OpenAI,
Groq, ...) answer ``/models`` with bare ``{id, object, owned_by}`` entries that
carry *no* pricing. The provider must not fabricate a price for those: it
resolves through native ``model_spec`` (Venice's bespoke schema) → litellm's
bundled cost map → the OpenRouter feed, and only when every source misses does
it import the model **disabled** with a warning rather than invent a number.
These tests drive that behaviour through the public ``fetch_models`` API. The
``/models`` HTTP call is faked at ``httpx.AsyncClient``; the OpenRouter feed is
patched at its source (``routstr.payment.models.async_fetch_openrouter_models``)
so the resolver's lazy import picks up the stub. litellm's real bundled cost map
is used unmocked — the DeepSeek rates it ships are the assertion's ground truth.
"""
from __future__ import annotations
import logging
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from routstr.upstream.generic import GenericUpstreamProvider
class _FakeResponse:
def __init__(self, payload: dict[str, Any]) -> None:
self._payload = payload
def raise_for_status(self) -> None:
return None
def json(self) -> dict[str, Any]:
return self._payload
class _FakeAsyncClient:
"""Stand-in for ``httpx.AsyncClient`` returning a canned ``/models`` body."""
def __init__(self, payload: dict[str, Any]) -> None:
self._payload = payload
async def __aenter__(self) -> "_FakeAsyncClient":
return self
async def __aexit__(self, *exc: object) -> bool:
return False
async def get(self, url: str, headers: dict[str, str] | None = None) -> _FakeResponse:
return _FakeResponse(self._payload)
def _patch_models_endpoint(payload: dict[str, Any]) -> Any:
"""Patch the provider's ``httpx.AsyncClient`` to serve ``payload``."""
return patch(
"routstr.upstream.generic.httpx.AsyncClient",
lambda *args, **kwargs: _FakeAsyncClient(payload),
)
def _model_by_id(models: list[Any], model_id: str) -> Any:
return next(m for m in models if m.id == model_id)
# ---------------------------------------------------------------------------
# native model_spec (Venice) — must keep resolving, and capture its metadata
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_native_model_spec_resolves_and_captures_metadata() -> None:
"""A Venice-shaped ``model_spec`` is authoritative: prices/context come
straight from it and vision capability becomes an image input modality."""
payload = {
"data": [
{
"id": "venice-llama",
"owned_by": "venice",
"model_spec": {
"name": "Venice Llama",
"availableContextTokens": 65536,
"pricing": {
"input": {"usd": 0.5},
"output": {"usd": 1.5},
},
"capabilities": {"supportsVision": True},
},
}
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "venice-llama")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(0.5 / 1_000_000)
assert model.pricing.completion == pytest.approx(1.5 / 1_000_000)
assert model.context_length == 65536
assert "image" in model.architecture.input_modalities
# A native price never needs the OpenRouter feed.
or_feed.assert_not_awaited()
# ---------------------------------------------------------------------------
# litellm rescue — the money-critical case (DeepSeek bare /models)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_bare_deepseek_resolves_via_litellm() -> None:
"""DeepSeek's ``/models`` carries no pricing. The old code fabricated
``$0.001`` + ctx 4096; the resolver must instead pull DeepSeek's real
rates from litellm's bundled cost map (``$0.28``/``$0.42`` per 1M, ctx
131072) and keep the model enabled."""
payload = {
"data": [
{"id": "deepseek-chat", "object": "model", "owned_by": "deepseek"},
]
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "deepseek-chat")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(2.8e-07)
assert model.pricing.completion == pytest.approx(4.2e-07)
assert model.context_length == 131072
# Richer metadata than the two base prices is captured too.
assert model.pricing.input_cache_read == pytest.approx(2.8e-08)
assert model.top_provider is not None
assert model.top_provider.max_completion_tokens == 8192
# litellm answered, so the OpenRouter feed is never consulted.
or_feed.assert_not_awaited()
# ---------------------------------------------------------------------------
# OpenRouter fallback — litellm misses, OR carries a full payload
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unknown_to_litellm_resolves_via_openrouter() -> None:
"""A model litellm has never heard of still resolves if the OpenRouter
feed lists it, pulling price + context from that entry."""
payload = {
"data": [
{"id": "exotic/model-9000", "object": "model", "owned_by": "exotic"},
]
}
or_entry = {
"id": "exotic/model-9000",
"name": "Exotic 9000",
"context_length": 65536,
"architecture": {
"modality": "text->text",
"input_modalities": ["text"],
"output_modalities": ["text"],
"tokenizer": "Other",
"instruct_type": None,
},
"pricing": {"prompt": "0.000005", "completion": "0.000010"},
"top_provider": {
"context_length": 65536,
"max_completion_tokens": 4096,
"is_moderated": False,
},
}
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[or_entry])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
model = _model_by_id(models, "exotic/model-9000")
assert model.enabled is True
assert model.pricing.prompt == pytest.approx(5e-06)
assert model.pricing.completion == pytest.approx(1e-05)
assert model.context_length == 65536
or_feed.assert_awaited()
@pytest.mark.asyncio
async def test_openrouter_feed_fetched_once_per_discovery() -> None:
"""Two models both missing litellm must share a single OpenRouter fetch —
the feed is not re-downloaded per model."""
payload = {
"data": [
{"id": "exotic/model-a", "object": "model", "owned_by": "exotic"},
{"id": "exotic/model-b", "object": "model", "owned_by": "exotic"},
]
}
or_feed = AsyncMock(
return_value=[
{
"id": "exotic/model-a",
"context_length": 8192,
"pricing": {"prompt": "0.000001", "completion": "0.000002"},
},
{
"id": "exotic/model-b",
"context_length": 8192,
"pricing": {"prompt": "0.000003", "completion": "0.000004"},
},
]
)
with _patch_models_endpoint(payload):
with patch("routstr.payment.models.async_fetch_openrouter_models", or_feed):
models = await GenericUpstreamProvider(base_url="http://x").fetch_models()
assert {m.id for m in models} == {"exotic/model-a", "exotic/model-b"}
assert or_feed.await_count == 1
# ---------------------------------------------------------------------------
# fail closed — no source resolves → disabled + warned, never fabricated
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unresolvable_model_fails_closed(
caplog: pytest.LogCaptureFixture,
) -> None:
"""When native, litellm and OpenRouter all miss, the model is imported
disabled with a warning naming it — and no price is invented (the old
``$0.001`` placeholder is gone)."""
payload = {
"data": [
{
"id": "nobody-has-priced-this-xyz",
"object": "model",
"owned_by": "mystery",
},
]
}
# routstr loggers set propagate=False, so caplog's root handler misses
# them; attach its handler to the provider logger directly.
gen_logger = logging.getLogger("routstr.upstream.generic")
gen_logger.addHandler(caplog.handler)
try:
with _patch_models_endpoint(payload):
or_feed = AsyncMock(return_value=[])
with patch(
"routstr.payment.models.async_fetch_openrouter_models", or_feed
):
models = await GenericUpstreamProvider(
base_url="http://x"
).fetch_models()
finally:
gen_logger.removeHandler(caplog.handler)
model = _model_by_id(models, "nobody-has-priced-this-xyz")
assert model.enabled is False
assert model.pricing.prompt == 0.0
assert model.pricing.completion == 0.0
assert any(
"nobody-has-priced-this-xyz" in rec.getMessage()
for rec in caplog.records
if rec.levelno >= logging.WARNING
)