diff --git a/routstr/upstream/__init__.py b/routstr/upstream/__init__.py index edac0020..85d094e8 100644 --- a/routstr/upstream/__init__.py +++ b/routstr/upstream/__init__.py @@ -13,6 +13,7 @@ from .ppqai import PPQAIUpstreamProvider from .routstr import RoutstrUpstreamProvider from .tinfoil import TinfoilUpstreamProvider from .typesafe import TypeSafeUpstreamProvider +from .venice import VeniceUpstreamProvider from .xai import XAIUpstreamProvider upstream_provider_classes: list[type[BaseUpstreamProvider]] = [ @@ -30,6 +31,7 @@ upstream_provider_classes: list[type[BaseUpstreamProvider]] = [ RoutstrUpstreamProvider, TinfoilUpstreamProvider, TypeSafeUpstreamProvider, + VeniceUpstreamProvider, XAIUpstreamProvider, ] """List of all upstream classes""" diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 01c52a03..5679aae3 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -2537,6 +2537,16 @@ class BaseUpstreamProvider: ) -> dict: return await messages_dispatch.aggregate_anthropic_events_to_message(iterator) + def adapt_messages_request(self, body: dict, model_obj: Model) -> str: + """Rewrite an allowlisted /v1/messages body for this upstream. + + Returns a suffix appended to the upstream model name, empty when the + provider needs none. Subclasses override this to express an Anthropic + feature the upstream spells differently; the base forwards the body + untouched. + """ + return "" + async def _dispatch_anthropic_messages( self, request_body: bytes | None, @@ -2551,6 +2561,7 @@ class BaseUpstreamProvider: api_key=self.api_key, provider_prefix=self.get_litellm_provider_prefix(), transform_model_name=self.transform_model_name, + adapt_request=lambda body: self.adapt_messages_request(body, model_obj), log_extra=log_extra, ) diff --git a/routstr/upstream/messages_dispatch.py b/routstr/upstream/messages_dispatch.py index 2856256c..488129a8 100644 --- a/routstr/upstream/messages_dispatch.py +++ b/routstr/upstream/messages_dispatch.py @@ -458,6 +458,7 @@ async def dispatch_anthropic_messages( api_key: str, provider_prefix: str, transform_model_name: Callable[[str], str], + adapt_request: Callable[[dict], str] | None = None, log_extra: dict[str, Any] | None = None, ) -> tuple[bool, Any, str | None]: """Call ``litellm.anthropic.messages.acreate`` and return @@ -465,6 +466,11 @@ async def dispatch_anthropic_messages( Shared by the bearer-key and x-cashu paths. Raises :class:`UpstreamError` on bad input or upstream failure. + + ``adapt_request`` is the provider's last word on the allowlisted body: it + may rewrite it in place and returns a suffix for the upstream model name, + which is how a provider expresses a feature litellm would otherwise + translate into a parameter the upstream rejects. """ if not request_body: raise UpstreamError("Missing request body for /v1/messages", status_code=400) @@ -499,13 +505,15 @@ async def dispatch_anthropic_messages( ) body = {k: v for k, v in body.items() if k in ALLOWED_MESSAGES_REQUEST_FIELDS} + model_suffix = adapt_request(body) if adapt_request else "" + # Convention: `model.id` is the canonical upstream model name; # `forwarded_model_id` is the public alias the internal API exposes # and echoes back to the client. requested_model = ( (model_obj.forwarded_model_id or model_obj.id) if model_obj else None ) - upstream_model = transform_model_name(model_obj.id) + upstream_model = f"{transform_model_name(model_obj.id)}{model_suffix}" litellm_model = f"{provider_prefix}{upstream_model}" kwargs: dict = { diff --git a/routstr/upstream/venice.py b/routstr/upstream/venice.py new file mode 100644 index 00000000..379a4a66 --- /dev/null +++ b/routstr/upstream/venice.py @@ -0,0 +1,312 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import httpx + +from ..core.exceptions import UpstreamError +from ..core.logging import get_logger +from ..payment.models import Architecture, Model, Pricing, TopProvider +from .base import BaseUpstreamProvider + +if TYPE_CHECKING: + from ..core.db import UpstreamProviderRow + +logger = get_logger(__name__) + +# ``GET /models`` defaults to ``type=text``, which is why a Venice account +# configured as a generic upstream never sees the rest of its catalog. +_MODELS_TYPE_PARAM = "all" + +# Families this proxy can both route and price. Image, audio, music and video +# are billed per clip or per second and return no usage object to settle +# against, so exposing them would hand out unpriced inference. +_SUPPORTED_TYPES = frozenset({"text", "embedding"}) + +# Venice prices text in USD per million tokens; Routstr prices per token. +_USD_PER_MILLION = 1_000_000.0 + +_ARCHITECTURES: dict[str, tuple[str, list[str], list[str]]] = { + "text": ("text->text", ["text"], ["text"]), + "embedding": ("text->embedding", ["text"], ["embedding"]), +} + +# Venice runs search itself and reports it back through ``venice_parameters``; +# it has no Anthropic-shaped server tool and rejects the ``web_search_options`` +# that litellm's Anthropic adapter derives from one. ``auto`` matches Anthropic +# semantics, where declaring the tool leaves the decision to the model. +# Citations are asked for because litellm's Anthropic response translation +# carries no ``venice_parameters``, so the inline ``^n^`` markers Venice writes +# into the text are the only way a caller sees that sources were used. +_WEB_SEARCH_SUFFIX = ":enable_web_search=auto&enable_web_citations=true" + +# Anthropic web-search constraints with no Venice equivalent. Honouring the +# request means enforcing them, so a request that sets one is refused rather +# than answered by a search that ignored it. ``max_uses`` is absent on purpose: +# ``auto`` runs at most one search per request, so any cap of 1 or more is +# already met, while domain filters and location would be silently ignored. +# Only ``max_uses: 0``, a request for no search at all, cannot be honoured. +_UNENFORCEABLE_WEB_SEARCH_KEYS = frozenset( + {"allowed_domains", "blocked_domains", "user_location"} +) + + +def _is_web_search_tool(tool: Any) -> bool: + """An Anthropic server-side web-search tool, by either of its markers. + + Matches litellm's own detection (``litellm/llms/anthropic/ + experimental_pass_through/adapters/transformation.py``), so every tool it + would turn into ``web_search_options`` is caught here first. + """ + if not isinstance(tool, dict): + return False + tool_type = tool.get("type") + return ( + isinstance(tool_type, str) and tool_type.startswith("web_search") + ) or tool.get("name") == "web_search" + + +def _usd(entry: Any) -> float | None: + """Read the USD leg of a Venice ``{usd, diem}`` price pair.""" + if isinstance(entry, dict): + value = entry.get("usd") + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + return None + + +class VeniceUpstreamProvider(BaseUpstreamProvider): + """Upstream provider for the Venice.ai API. + + Venice publishes a complete price book on its own catalog, so models are + built from that rather than matched against OpenRouter, which has never + heard of most of Venice's catalog. + """ + + provider_type = "venice" + default_base_url = "https://api.venice.ai/api/v1" + platform_url = "https://venice.ai/settings/api" + + def __init__(self, api_key: str, provider_fee: float = 1.01): + super().__init__( + base_url=self.default_base_url, api_key=api_key, provider_fee=provider_fee + ) + + @classmethod + def _build_from_row( + cls, provider_row: "UpstreamProviderRow" + ) -> "VeniceUpstreamProvider": + return cls( + api_key=provider_row.api_key, + provider_fee=provider_row.provider_fee, + ) + + @classmethod + def get_provider_metadata(cls) -> dict[str, object]: + return { + "id": cls.provider_type, + "name": "Venice AI", + "default_base_url": cls.default_base_url, + "fixed_base_url": True, + "platform_url": cls.platform_url, + } + + def transform_model_name(self, model_id: str) -> str: + return model_id.removeprefix("venice/") + + def adapt_messages_request(self, body: dict, model_obj: Model) -> str: + """Trade an Anthropic web-search tool for Venice's own search switch. + + Left in the body, litellm's Anthropic adapter rewrites the tool into a + top-level ``web_search_options``, which Venice answers with a 400. The + tool is lifted out here and the same intent re-expressed as a model + feature suffix, the one form of ``venice_parameters`` that survives + that adapter. + """ + tools = body.get("tools") + if not isinstance(tools, list): + return "" + search_tools = [tool for tool in tools if _is_web_search_tool(tool)] + if not search_tools: + return "" + + # A key carrying null or an empty list states no constraint, so it is + # read as absent rather than refused. ``auto`` runs at most one search, + # so only an integer ``max_uses`` of one or more is known to be met. + unenforceable = sorted( + { + key + for tool in search_tools + for key, value in tool.items() + if ( + key in _UNENFORCEABLE_WEB_SEARCH_KEYS + and value is not None + and value != [] + ) + or ( + key == "max_uses" + and value is not None + and not ( + isinstance(value, int) + and not isinstance(value, bool) + and value >= 1 + ) + ) + } + ) + if unenforceable: + raise UpstreamError( + "Venice web search cannot honour these Anthropic web_search " + f"options: {', '.join(unenforceable)}", + status_code=400, + code="UNSUPPORTED_WEB_SEARCH_OPTION", + details={"unsupported_options": unenforceable}, + ) + + tool_choice = body.get("tool_choice") + if isinstance(tool_choice, dict) and tool_choice.get("name") == "web_search": + raise UpstreamError( + "Venice web search cannot be forced through tool_choice; it is " + "decided by the model", + status_code=400, + code="UNSUPPORTED_WEB_SEARCH_OPTION", + details={"unsupported_options": ["tool_choice"]}, + ) + + remaining = [tool for tool in tools if not _is_web_search_tool(tool)] + if remaining: + # A caller's ``tool_choice: any`` is kept and litellm maps it to + # OpenAI ``required``, so one of the remaining function tools must + # now be called where Anthropic would have let a search satisfy it. + # Deliberate: OpenRouter never rewrites tool_choice for web search + # either, and guessing an alternative would change caller intent. + body["tools"] = remaining + else: + body.pop("tools", None) + # tool_choice without tools is rejected by OpenAI-shaped upstreams. + body.pop("tool_choice", None) + + return _WEB_SEARCH_SUFFIX + + async def _fetch_provider_models(self) -> dict: + url = f"{self.base_url.rstrip('/')}/models" + headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else None + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.get( + url, params={"type": _MODELS_TYPE_PARAM}, headers=headers + ) + response.raise_for_status() + return response.json() + + async def fetch_models(self) -> list[Model]: + try: + payload = await self._fetch_provider_models() + except Exception as e: + logger.error( + "Error fetching Venice models", + extra={"error": str(e), "error_type": type(e).__name__}, + ) + return [] + + models: list[Model] = [] + skipped: list[str] = [] + for entry in payload.get("data", []): + if not isinstance(entry, dict): + continue + try: + model = self._parse_model(entry) + except Exception as e: + logger.warning( + "Failed to parse Venice model", + extra={ + "model_id": entry.get("id", "unknown"), + "error": str(e), + "error_type": type(e).__name__, + }, + ) + continue + if model is None: + skipped.append(str(entry.get("id", "unknown"))) + continue + models.append(model) + + if skipped: + logger.debug( + f"({len(skipped)}) Venice models skipped as unsupported or unpriced", + extra={"skipped_models": skipped}, + ) + return models + + def _parse_model(self, entry: dict[str, Any]) -> Model | None: + model_type = entry.get("type") + model_id = entry.get("id") + spec = entry.get("model_spec") + if not model_id or model_type not in _SUPPORTED_TYPES: + return None + if not isinstance(spec, dict) or spec.get("offline"): + return None + + pricing = self._parse_pricing(spec.get("pricing"), str(model_type)) + if pricing is None: + return None + + modality, input_modalities, output_modalities = _ARCHITECTURES[str(model_type)] + capabilities = spec.get("capabilities") + if ( + model_type == "text" + and isinstance(capabilities, dict) + and capabilities.get("supportsVision") + ): + input_modalities = [*input_modalities, "image"] + modality = "text+image->text" + + context_length = spec.get("availableContextTokens") + max_completion_tokens = spec.get("maxCompletionTokens") + name = spec.get("name") or str(model_id) + + return Model( + id=str(model_id), + name=str(name), + created=int(entry.get("created") or 0), + description=str(spec.get("description") or f"Venice {model_type} model"), + context_length=int(context_length) if context_length else 0, + architecture=Architecture( + modality=modality, + input_modalities=input_modalities, + output_modalities=output_modalities, + tokenizer="Unknown", + instruct_type=None, + ), + pricing=pricing, + top_provider=TopProvider( + context_length=int(context_length) if context_length else None, + max_completion_tokens=int(max_completion_tokens) + if max_completion_tokens + else None, + ), + ) + + def _parse_pricing(self, raw: Any, model_type: str) -> Pricing | None: + if not isinstance(raw, dict): + return None + + # The ``extended`` tier some models charge past a context threshold is + # ignored: billing it would overcharge every request staying under it. + input_usd = _usd(raw.get("input")) + output_usd = _usd(raw.get("output")) + # Embeddings produce no completion tokens, so only they may omit an + # output price. Anywhere else a missing or all-zero price would serve + # completions free and a negative one would credit the caller, the + # same guards ``generic.py`` applies to this price book. + if output_usd is None and model_type == "embedding": + output_usd = 0.0 + if input_usd is None or output_usd is None: + return None + if input_usd < 0 or output_usd < 0 or (input_usd == 0 and output_usd == 0): + return None + return Pricing( + prompt=input_usd / _USD_PER_MILLION, + completion=output_usd / _USD_PER_MILLION, + input_cache_read=(_usd(raw.get("cache_input")) or 0.0) / _USD_PER_MILLION, + input_cache_write=(_usd(raw.get("cache_write")) or 0.0) / _USD_PER_MILLION, + ) diff --git a/tests/integration/test_venice_web_search_wire.py b/tests/integration/test_venice_web_search_wire.py new file mode 100644 index 00000000..2636af1f --- /dev/null +++ b/tests/integration/test_venice_web_search_wire.py @@ -0,0 +1,131 @@ +"""What Routstr actually puts on the wire for a Venice web-search request. + +The unit tests stop at the kwargs handed to litellm. Everything that produced +the reported ``400 Unrecognized key(s) in object: 'web_search_options'`` +happened *after* that point, inside litellm's Anthropic adapter, so this test +runs the whole dispatch against a loopback OpenAI-compatible server and reads +the bytes Venice would have received. +""" + +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any, Iterator + +import pytest + +from routstr.payment.models import Architecture, Model, Pricing +from routstr.upstream.litellm_routing import configure_litellm +from routstr.upstream.venice import VeniceUpstreamProvider + +_CHUNKS = [ + { + "id": "chatcmpl-1", + "object": "chat.completion.chunk", + "created": 0, + "model": "deepseek-v4-flash-0731", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}], + }, + { + "id": "chatcmpl-1", + "object": "chat.completion.chunk", + "created": 0, + "model": "deepseek-v4-flash-0731", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7}, + }, +] + + +@pytest.fixture +def upstream() -> Iterator[tuple[str, dict[str, Any]]]: + """A loopback stand-in for ``api.venice.ai`` that records one request.""" + captured: dict[str, Any] = {} + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 - http.server's spelling + length = int(self.headers.get("Content-Length", 0)) + captured["path"] = self.path + captured["body"] = json.loads(self.rfile.read(length)) + + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + for chunk in _CHUNKS: + self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode()) + self.wfile.write(b"data: [DONE]\n\n") + + def log_message(self, *args: Any) -> None: + return None + + server = HTTPServer(("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_address[1]}/v1", captured + finally: + server.shutdown() + thread.join(timeout=5) + + +def _model() -> Model: + return Model( + id="deepseek-v4-flash-0731", + name="deepseek-v4-flash-0731", + created=0, + description="", + context_length=8192, + architecture=Architecture( + modality="text->text", + input_modalities=["text"], + output_modalities=["text"], + tokenizer="Unknown", + instruct_type=None, + ), + pricing=Pricing(prompt=0.0, completion=0.0), + ) + + +@pytest.mark.asyncio +async def test_web_search_request_reaches_venice_in_its_own_shape( + upstream: tuple[str, dict[str, Any]], +) -> None: + base_url, captured = upstream + # The app applies this at startup; without it litellm posts the Anthropic + # body to /responses, which Venice serves only in alpha. + configure_litellm() + + provider = VeniceUpstreamProvider(api_key="sk-test") + provider.base_url = base_url + + await provider._dispatch_anthropic_messages( + request_body=json.dumps( + { + "model": "venice/deepseek-v4-flash-0731", + "messages": [{"role": "user", "content": "what shipped today?"}], + "max_tokens": 64, + "stream": True, + "tools": [ + {"type": "web_search_20250305", "name": "web_search"}, + { + "name": "lookup", + "description": "Look something up", + "input_schema": {"type": "object", "properties": {}}, + }, + ], + } + ).encode(), + model_obj=_model(), + ) + + body = captured["body"] + assert captured["path"] == "/v1/chat/completions" + # The reported 400, at the only place it could be observed. + assert "web_search_options" not in body + assert body["model"] == ( + "deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true" + ) + # The function tool still travels, in OpenAI's shape. + assert [tool["function"]["name"] for tool in body["tools"]] == ["lookup"] diff --git a/tests/unit/test_upstream_venice.py b/tests/unit/test_upstream_venice.py new file mode 100644 index 00000000..742c38ae --- /dev/null +++ b/tests/unit/test_upstream_venice.py @@ -0,0 +1,293 @@ +"""Unit tests for ``VeniceUpstreamProvider.fetch_models``. + +Venice answers ``/models`` with only its text catalog unless ``type`` is +passed, which is why the same account configured as a generic upstream sees a +different catalog. These tests pin that query parameter, the per-token pricing +shape, and the families dropped as unpriceable. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import patch + +import pytest + +from routstr.upstream.venice import VeniceUpstreamProvider + + +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: + def __init__(self, payload: dict[str, Any], calls: list[dict[str, Any]]) -> None: + self._payload = payload + self._calls = calls + + async def __aenter__(self) -> "_FakeAsyncClient": + return self + + async def __aexit__(self, *_: object) -> None: + return None + + async def get( + self, + url: str, + params: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, + ) -> _FakeResponse: + self._calls.append({"url": url, "params": params, "headers": headers}) + return _FakeResponse(self._payload) + + +CATALOG: dict[str, Any] = { + "object": "list", + "data": [ + { + "id": "venice-uncensored-1-2", + "type": "text", + "created": 1727966436, + "model_spec": { + "name": "Venice Uncensored 1.2", + "availableContextTokens": 128000, + "maxCompletionTokens": 8192, + "capabilities": {"supportsVision": True}, + "pricing": { + "input": {"usd": 0.2, "diem": 0.2}, + "output": {"usd": 0.9, "diem": 0.9}, + "cache_input": {"usd": 0.02, "diem": 0.02}, + "cache_write": {"usd": 0.25, "diem": 0.25}, + }, + }, + }, + { + "id": "text-embedding-bge-m3", + "type": "embedding", + "created": 1727966436, + "model_spec": { + "name": "BGE m3", + "availableContextTokens": 8192, + "pricing": {"input": {"usd": 0.01, "diem": 0.01}}, + }, + }, + { + "id": "unpriced-text", + "type": "text", + "created": 1727966436, + "model_spec": {"name": "Unpriced", "pricing": {}}, + }, + { + "id": "offline-model", + "type": "text", + "created": 1727966436, + "model_spec": { + "name": "Offline", + "offline": True, + "pricing": {"input": {"usd": 0.2, "diem": 0.2}}, + }, + }, + { + "id": "venice-sd35", + "type": "image", + "created": 1727966436, + "model_spec": { + "name": "Venice SD35", + "pricing": {"generation": {"usd": 0.01, "diem": 0.01}}, + }, + }, + { + "id": "flux-2-max-edit", + "type": "inpaint", + "created": 1727966436, + "model_spec": { + "name": "FLUX.2 Max Edit", + "pricing": {"inpaint": {"usd": 0.12, "diem": 0.12}}, + }, + }, + { + "id": "tts-kokoro", + "type": "tts", + "created": 1727966436, + "model_spec": { + "name": "Kokoro", + "pricing": {"input": {"usd": 3.5, "diem": 3.5}}, + }, + }, + { + "id": "unpriced-video", + "type": "video", + "created": 1727966436, + "model_spec": {"name": "Video"}, + }, + ], +} + + +def _fetch(payload: dict[str, Any] = CATALOG) -> tuple[list[Any], list[dict[str, Any]]]: + import asyncio + + calls: list[dict[str, Any]] = [] + provider = VeniceUpstreamProvider(api_key="sk-test") + with patch( + "routstr.upstream.venice.httpx.AsyncClient", + lambda *a, **kw: _FakeAsyncClient(payload, calls), + ): + models = asyncio.run(provider.fetch_models()) + return models, calls + + +def test_requests_every_model_family() -> None: + _, calls = _fetch() + assert calls[0]["params"] == {"type": "all"} + assert calls[0]["url"] == "https://api.venice.ai/api/v1/models" + assert calls[0]["headers"] == {"Authorization": "Bearer sk-test"} + + +def test_text_pricing_is_per_token() -> None: + models, _ = _fetch() + model = next(m for m in models if m.id == "venice-uncensored-1-2") + assert model.pricing.prompt == pytest.approx(0.2 / 1_000_000) + assert model.pricing.completion == pytest.approx(0.9 / 1_000_000) + assert model.pricing.input_cache_read == pytest.approx(0.02 / 1_000_000) + assert model.pricing.input_cache_write == pytest.approx(0.25 / 1_000_000) + assert model.context_length == 128000 + assert model.top_provider is not None + assert model.top_provider.max_completion_tokens == 8192 + assert model.architecture.input_modalities == ["text", "image"] + assert model.architecture.modality == "text+image->text" + + +def test_embedding_models_are_listed() -> None: + models, _ = _fetch() + model = next(m for m in models if m.id == "text-embedding-bge-m3") + assert model.architecture.output_modalities == ["embedding"] + assert model.pricing.prompt == pytest.approx(0.01 / 1_000_000) + assert model.pricing.completion == 0.0 + + +def test_families_billed_per_clip_are_dropped() -> None: + """Image, audio and video return no usage to settle against, so listing + them here would hand out inference this provider cannot price.""" + models, _ = _fetch() + ids = {m.id for m in models} + assert "venice-sd35" not in ids + assert "flux-2-max-edit" not in ids + assert "tts-kokoro" not in ids + assert "unpriced-video" not in ids + + +def test_offline_and_unpriced_models_are_dropped() -> None: + models, _ = _fetch() + ids = {m.id for m in models} + assert "offline-model" not in ids + assert "unpriced-text" not in ids + + +def _priced_entry(model_id: str, model_type: str, pricing: dict[str, Any]) -> dict: + return { + "id": model_id, + "type": model_type, + "created": 1727966436, + "model_spec": {"name": model_id, "pricing": pricing}, + } + + +@pytest.mark.parametrize( + "pricing", + [ + pytest.param({"input": {"usd": 0.2, "diem": 0.2}}, id="missing-output"), + pytest.param( + {"input": {"usd": 0.0, "diem": 0.0}, "output": {"usd": 0.0, "diem": 0.0}}, + id="both-zero", + ), + pytest.param( + {"input": {"usd": -0.2, "diem": 0.2}, "output": {"usd": 0.9, "diem": 0.9}}, + id="negative-input", + ), + pytest.param( + {"input": {"usd": 0.2, "diem": 0.2}, "output": {"usd": -0.9, "diem": 0.9}}, + id="negative-output", + ), + ], +) +def test_text_models_that_would_bill_free_or_negative_are_dropped( + pricing: dict[str, Any], +) -> None: + models, _ = _fetch( + {"object": "list", "data": [_priced_entry("bad-text", "text", pricing)]} + ) + assert models == [] + + +def test_embedding_with_only_an_input_price_is_listed() -> None: + models, _ = _fetch( + { + "object": "list", + "data": [ + _priced_entry("emb", "embedding", {"input": {"usd": 0.05, "diem": 0}}) + ], + } + ) + assert [m.id for m in models] == ["emb"] + assert models[0].pricing.prompt == pytest.approx(0.05 / 1_000_000) + assert models[0].pricing.completion == 0.0 + + +def test_embedding_with_a_negative_price_is_dropped() -> None: + models, _ = _fetch( + { + "object": "list", + "data": [ + _priced_entry("emb", "embedding", {"input": {"usd": -0.05, "diem": 0}}) + ], + } + ) + assert models == [] + + +def test_text_model_with_one_zero_price_is_listed() -> None: + """Only both-zero is free; a free prompt with a paid completion is priced.""" + pricing = {"input": {"usd": 0.0, "diem": 0}, "output": {"usd": 0.9, "diem": 0}} + models, _ = _fetch( + {"object": "list", "data": [_priced_entry("t", "text", pricing)]} + ) + assert [m.id for m in models] == ["t"] + assert models[0].pricing.completion == pytest.approx(0.9 / 1_000_000) + + +def test_model_name_drops_the_venice_prefix() -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + assert provider.transform_model_name("venice/venice-uncensored-1-2") == ( + "venice-uncensored-1-2" + ) + assert provider.transform_model_name("venice-uncensored-1-2") == ( + "venice-uncensored-1-2" + ) + + +def test_provider_metadata_pins_the_base_url() -> None: + metadata = VeniceUpstreamProvider.get_provider_metadata() + assert metadata["id"] == "venice" + assert metadata["default_base_url"] == "https://api.venice.ai/api/v1" + assert metadata["fixed_base_url"] is True + + +def test_fetch_returns_empty_on_upstream_failure() -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + + with patch.object( + VeniceUpstreamProvider, + "_fetch_provider_models", + side_effect=RuntimeError("boom"), + ): + import asyncio + + assert asyncio.run(provider.fetch_models()) == [] diff --git a/tests/unit/test_venice_web_search.py b/tests/unit/test_venice_web_search.py new file mode 100644 index 00000000..a836e956 --- /dev/null +++ b/tests/unit/test_venice_web_search.py @@ -0,0 +1,281 @@ +"""Venice web search over ``/v1/messages``. + +litellm's Anthropic adapter rewrites an Anthropic server-side web-search tool +into a top-level ``web_search_options``, which Venice rejects with +``400 Unrecognized key(s) in object: 'web_search_options'``. These tests pin +the trade: the tool is lifted out of the body and the same intent re-expressed +as a Venice model feature suffix. +""" + +from __future__ import annotations + +import json +from typing import Any, AsyncIterator +from unittest.mock import AsyncMock, patch + +import pytest + +from routstr.core.exceptions import UpstreamError +from routstr.payment.models import Architecture, Model, Pricing +from routstr.upstream.base import BaseUpstreamProvider +from routstr.upstream.venice import VeniceUpstreamProvider + +WEB_SEARCH_TOOL = {"type": "web_search_20250305", "name": "web_search"} +FUNCTION_TOOL = { + "name": "lookup", + "description": "Look something up", + "input_schema": {"type": "object", "properties": {}}, +} + + +def _model(model_id: str = "deepseek-v4-flash-0731") -> Model: + return Model( + id=model_id, + name=model_id, + created=0, + description="", + context_length=8192, + architecture=Architecture( + modality="text->text", + input_modalities=["text"], + output_modalities=["text"], + tokenizer="Unknown", + instruct_type=None, + ), + pricing=Pricing(prompt=0.0, completion=0.0), + ) + + +def _body(**extra: Any) -> dict[str, Any]: + return { + "messages": [{"role": "user", "content": "what shipped today?"}], + "max_tokens": 64, + **extra, + } + + +async def _dispatch(provider: BaseUpstreamProvider, body: dict[str, Any]) -> dict: + """Run the real dispatcher, capturing the kwargs litellm would receive.""" + captured: dict[str, Any] = {} + + async def empty_iter() -> AsyncIterator[dict]: + if False: + yield {} + + async def fake_acreate(**kwargs: Any) -> AsyncIterator[dict]: + captured.update(kwargs) + return empty_iter() + + with patch( + "litellm.anthropic.messages.acreate", + new=AsyncMock(side_effect=fake_acreate), + ): + await provider._dispatch_anthropic_messages( + request_body=json.dumps( + {"model": "venice/x", "stream": True, **body} + ).encode(), + model_obj=_model(), + ) + return captured + + +@pytest.mark.asyncio +async def test_web_search_tool_never_reaches_venice_as_web_search_options() -> None: + """The reported 400: the derived parameter must not be sent at all.""" + provider = VeniceUpstreamProvider(api_key="sk-test") + + kwargs = await _dispatch(provider, _body(tools=[WEB_SEARCH_TOOL])) + + assert "web_search_options" not in kwargs + assert "tools" not in kwargs + assert kwargs["model"] == ( + "openai/deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true" + ) + assert kwargs["api_base"] == "https://api.venice.ai/api/v1" + + +@pytest.mark.asyncio +async def test_function_tools_survive_alongside_web_search() -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + + kwargs = await _dispatch( + provider, + _body( + tools=[WEB_SEARCH_TOOL, FUNCTION_TOOL], + tool_choice={"type": "tool", "name": "lookup"}, + ), + ) + + assert kwargs["tools"] == [FUNCTION_TOOL] + assert kwargs["tool_choice"] == {"type": "tool", "name": "lookup"} + assert "web_search_options" not in kwargs + assert kwargs["model"].endswith(":enable_web_search=auto&enable_web_citations=true") + + +@pytest.mark.asyncio +async def test_requests_without_web_search_are_untouched() -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + + kwargs = await _dispatch(provider, _body(tools=[FUNCTION_TOOL])) + + assert kwargs["model"] == "openai/deepseek-v4-flash-0731" + assert kwargs["tools"] == [FUNCTION_TOOL] + + +@pytest.mark.asyncio +async def test_other_providers_keep_their_existing_behaviour() -> None: + """The base hook is a no-op, so no non-Venice upstream changes shape.""" + provider = BaseUpstreamProvider(base_url="http://test", api_key="k") + + kwargs = await _dispatch(provider, _body(tools=[WEB_SEARCH_TOOL])) + + assert kwargs["model"] == "openai/deepseek-v4-flash-0731" + assert kwargs["tools"] == [WEB_SEARCH_TOOL] + + +@pytest.mark.parametrize( + "tool", + [ + { + "type": "web_search_20250305", + "name": "web_search", + "allowed_domains": ["example.com"], + }, + {"type": "web_search_20250305", "name": "web_search", "blocked_domains": ["x"]}, + { + "type": "web_search_20250305", + "name": "web_search", + "user_location": {"type": "approximate", "country": "DE"}, + }, + ], +) +def test_constraints_venice_cannot_enforce_are_refused(tool: dict[str, Any]) -> None: + """Better an explicit 400 than a search that quietly ignored the limit.""" + provider = VeniceUpstreamProvider(api_key="sk-test") + + with pytest.raises(UpstreamError) as excinfo: + provider.adapt_messages_request(_body(tools=[tool]), _model()) + + assert excinfo.value.status_code == 400 + assert excinfo.value.code == "UNSUPPORTED_WEB_SEARCH_OPTION" + + +@pytest.mark.parametrize( + "tool", + [ + {"type": "web_search_20250305", "name": "web_search", "max_uses": None}, + {"type": "web_search_20250305", "name": "web_search", "allowed_domains": []}, + ], +) +def test_constraint_keys_stating_nothing_are_read_as_absent( + tool: dict[str, Any], +) -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + + assert provider.adapt_messages_request(_body(tools=[tool]), _model()) != "" + + +def test_forcing_web_search_through_tool_choice_is_refused() -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + body = _body( + tools=[WEB_SEARCH_TOOL], + tool_choice={"type": "tool", "name": "web_search"}, + ) + + with pytest.raises(UpstreamError) as excinfo: + provider.adapt_messages_request(body, _model()) + + assert excinfo.value.status_code == 400 + assert excinfo.value.code == "UNSUPPORTED_WEB_SEARCH_OPTION" + assert excinfo.value.details == {"unsupported_options": ["tool_choice"]} + + +def test_web_search_only_request_drops_tool_choice() -> None: + """Without tools left, a surviving tool_choice is rejected upstream.""" + provider = VeniceUpstreamProvider(api_key="sk-test") + body = _body(tools=[WEB_SEARCH_TOOL], tool_choice={"type": "auto"}) + + provider.adapt_messages_request(body, _model()) + + assert "tools" not in body + assert "tool_choice" not in body + + +@pytest.mark.asyncio +async def test_claude_code_web_search_tool_is_accepted() -> None: + """Claude Code always sends ``max_uses: 8``; Venice's single ``auto`` + search already stays under any cap of one or more.""" + provider = VeniceUpstreamProvider(api_key="sk-test") + tool = { + "type": "web_search_20250305", + "name": "web_search", + "allowed_domains": None, + "blocked_domains": None, + "max_uses": 8, + } + + kwargs = await _dispatch(provider, _body(tools=[tool])) + + assert "web_search_options" not in kwargs + assert "tools" not in kwargs + assert kwargs["model"] == ( + "openai/deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true" + ) + + +@pytest.mark.parametrize("max_uses", [1, None]) +def test_max_uses_of_one_or_absent_is_accepted(max_uses: Any) -> None: + provider = VeniceUpstreamProvider(api_key="sk-test") + tool = {"type": "web_search_20250305", "name": "web_search", "max_uses": max_uses} + + assert provider.adapt_messages_request(_body(tools=[tool]), _model()) != "" + + +@pytest.mark.parametrize("max_uses", [0, -1, 1.5, True, "0", "8"]) +def test_max_uses_other_than_a_positive_integer_is_refused(max_uses: Any) -> None: + """``auto`` may still search, so a cap below one cannot be met, and a + malformed cap cannot be shown to be met.""" + provider = VeniceUpstreamProvider(api_key="sk-test") + tool = {"type": "web_search_20250305", "name": "web_search", "max_uses": max_uses} + + with pytest.raises(UpstreamError) as excinfo: + provider.adapt_messages_request(_body(tools=[tool]), _model()) + + assert excinfo.value.status_code == 400 + assert excinfo.value.code == "UNSUPPORTED_WEB_SEARCH_OPTION" + assert excinfo.value.details == {"unsupported_options": ["max_uses"]} + + +def test_tool_named_web_search_without_the_type_marker_is_caught() -> None: + """litellm matches on either marker, so this one would also be rewritten.""" + provider = VeniceUpstreamProvider(api_key="sk-test") + body = _body(tools=[{"name": "web_search"}]) + + assert provider.adapt_messages_request(body, _model()) != "" + assert "tools" not in body + + +def test_litellm_adapter_derives_no_web_search_options_from_the_adapted_body() -> None: + """The fix at its cause: run the real litellm translation over the body + this provider produces and assert the rejected key is never derived.""" + from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import ( # noqa: E501 + LiteLLMAnthropicMessagesAdapter, + ) + + provider = VeniceUpstreamProvider(api_key="sk-test") + adapter = LiteLLMAnthropicMessagesAdapter() # type: ignore[no-untyped-call] + body = _body(tools=[WEB_SEARCH_TOOL, FUNCTION_TOOL]) + + def translate(request: dict[str, Any]) -> dict: + # litellm types the request as a TypedDict; these bodies are built + # from client JSON, so they are plain dicts at this seam. + translated, _ = adapter.translate_anthropic_to_openai(request) # type: ignore[arg-type] + return dict(translated) + + # Unadapted, litellm derives the parameter Venice rejects. + before = translate({"model": "m", **_body(tools=[WEB_SEARCH_TOOL])}) + assert "web_search_options" in before + + provider.adapt_messages_request(body, _model()) + + assert "web_search_options" not in translate({"model": "m", **body})