fix(databricks): call the ai_decide AI Function route for decisions and the auto-router (#45722)

* fix(databricks): call the ai_decide AI Function route for decisions and the auto-router

Databricks serves ai_decide at POST <host>/api/2.0/ai-functions/ai-decide, not
on a serving endpoint's /invocations route. The decisions provider and the
auto-router classifier now accept only the ai_decide model, post to that route
without a model field, unwrap the response envelope, and map a noul answer's
probability to the System One noul field.

* test(databricks): bound the megabyte model name test by scaling, not wall time

* test(databricks): assert the whole OpenAI-format decision and floor the liveliness bound

* test(decisions): move the Databricks OpenAI-format wire row to the ai_decide route

* fix(databricks): keep serving endpoint decisions beside the ai_decide route

* test(databricks): assert the whole classifier request body in the test routing case

* test(databricks): correct System One spend call type

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: mateo <mateo@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-10 23:58:16 -07:00 • committed by GitHub
parent 246c0e126b
commit ca0ff24726
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 761 additions and 191 deletions

View file

@ -4,18 +4,63 @@ from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Final
from pydantic import TypeAdapter, ValidationError
from litellm.llms.base_llm.decisions.transformation import BaseDecisionsConfig
from litellm.types.decisions import DecisionsIRRequest, UnsupportedDecisionsRequest
DATABRICKS_AI_DECIDE_MODEL: Final = "ai_decide"
_AI_DECIDE_PATH: Final = "/api/2.0/ai-functions/ai-decide"
_SERVING_ENDPOINT_NAME: Final = re.compile(r"[A-Za-z0-9_-][A-Za-z0-9._-]*")
_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
def validate_serving_endpoint_name(name: str) -> str:
if _SERVING_ENDPOINT_NAME.fullmatch(name) is None:
raise ValueError(
f"Databricks serving endpoint name {name!r} must be the bare endpoint name (letters, digits, '-', '_' "
"and '.', with no '/', '?', '#', spaces, or a leading '.'), e.g. databricks-openjev-qwen35-4b"
)
return name
def is_ai_decide_model(model: str) -> bool:
return model == DATABRICKS_AI_DECIDE_MODEL
def validate_databricks_decisions_model(model: str) -> str:
if is_ai_decide_model(model) or _SERVING_ENDPOINT_NAME.fullmatch(model) is not None:
return model
raise ValueError(
f"Databricks decisions model must be {DATABRICKS_AI_DECIDE_MODEL!r} (the ai_decide AI Function) or a bare "
"serving endpoint name (letters, digits, '-', '_' and '.', with no '/', '?', '#', spaces, or a leading '.'), "
"e.g. databricks-openjev-qwen35-4b"
)
def databricks_workspace_host(api_base: str) -> str:
return api_base.rstrip("/").removesuffix("/serving-endpoints")
def _mapping(value: object) -> Mapping[str, object] | None:
try:
return _MAPPING_ADAPTER.validate_python(value)
except ValidationError:
return None
def _systemone_answer(answer: object) -> object:
fields: Final = _mapping(answer)
if fields is None or fields.get("type") != "noul" or "probability" not in fields:
return answer
return MappingProxyType(
{**{key: value for key, value in fields.items() if key != "probability"}, "noul": fields["probability"]}
)
def ai_decide_response(payload: Mapping[str, object]) -> Mapping[str, object]:
response: Final = _mapping(payload.get("response"))
answers: Final = None if response is None else _mapping(response.get("answers"))
if response is None or answers is None:
return payload
return MappingProxyType(
{
**{key: value for key, value in payload.items() if key != "response"},
**response,
"answers": MappingProxyType({key: _systemone_answer(answer) for key, answer in answers.items()}),
}
)
@dataclass(frozen=True, slots=True)
@ -35,13 +80,30 @@ class DatabricksDecisionsConfig(BaseDecisionsConfig):
)
def canonical_model(self, model: str) -> str:
return validate_serving_endpoint_name(model)
return validate_databricks_decisions_model(model)
def get_complete_url(self, api_base: str, model: str) -> str:
return f"{api_base.rstrip('/')}/{validate_serving_endpoint_name(model)}/invocations"
if is_ai_decide_model(model):
return f"{databricks_workspace_host(api_base)}{_AI_DECIDE_PATH}"
return f"{api_base.rstrip('/')}/{validate_databricks_decisions_model(model)}/invocations"
def transform_decisions_request(
self,
model: str,
request: DecisionsIRRequest,
custom_llm_provider: str,
) -> Mapping[str, object] | UnsupportedDecisionsRequest:
body: Final = super().transform_decisions_request(model, request, custom_llm_provider)
if isinstance(body, UnsupportedDecisionsRequest) or not is_ai_decide_model(model):
return body
return MappingProxyType({key: value for key, value in body.items() if key != "model"})
def unwrap_response(self, payload: object) -> object:
fields: Final = _mapping(payload)
return payload if fields is None else ai_decide_response(fields)
def classifier_response(self, body: Mapping[str, object], requested_model: str) -> Mapping[str, object]:
return MappingProxyType({**body, "model": requested_model})
return MappingProxyType({**ai_decide_response(body), "model": requested_model})
def connection(self, api_base: str | None, api_key: str | None) -> DatabricksDecisionsConnection:
base: Final = self.resolve_api_base(api_base)

View file

@ -21177,6 +21177,18 @@
"output_cost_per_token": 0.0,
"source": "https://docs.databricks.com/aws/en/machine-learning/foundation-model-apis/supported-models"
},
"databricks/ai_decide": {
"cache_creation_input_token_cost": 0.0,
"cache_read_input_token_cost": 0.0,
"input_cost_per_token": 0.0,
"litellm_provider": "databricks",
"metadata": {
"notes": "The ai_decide AI Function (Beta), called through POST /api/2.0/ai-functions/ai-decide. Its response carries no token usage, so LiteLLM records no spend for it."
},
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.databricks.com/api/ai-functions/v1/ai-decide"
},
"dataforseo/search": {
"input_cost_per_query": 0.003,
"litellm_provider": "dataforseo",

View file

@ -769,14 +769,17 @@ class OpenSourceClassifierConfig(LiteLLMBaseModel):
_ = validate_oss_api_base(self.provider, self.api_base)
return self
if self.provider == "databricks":
from litellm.llms.databricks.decisions.transformation import validate_serving_endpoint_name
from litellm.llms.databricks.decisions.transformation import (
DATABRICKS_AI_DECIDE_MODEL,
validate_databricks_decisions_model,
)
if "model" not in self.model_fields_set:
raise ValueError(
"opensource_classifier_config.model is required for provider 'databricks': the serving endpoint "
"name, e.g. databricks-openjev-qwen35-4b"
"opensource_classifier_config.model is required for provider 'databricks': "
f"{DATABRICKS_AI_DECIDE_MODEL!r} or a serving endpoint name, e.g. databricks-openjev-qwen35-4b"
)
_ = validate_serving_endpoint_name(self.model)
_ = validate_databricks_decisions_model(self.model)
if self.api_base is not None and self.api_key is None:
raise ValueError(
"opensource_classifier_config.api_base requires opensource_classifier_config.api_key: "

View file

@ -19,7 +19,7 @@ from litellm.litellm_core_utils.internal_call_metadata import (
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.llms.databricks.decisions.transformation import DATABRICKS_DECISIONS_CONFIG
from litellm.llms.databricks.decisions.transformation import DATABRICKS_DECISIONS_CONFIG, is_ai_decide_model
from litellm.llms.laya.common_utils import laya_response_model
from litellm.llms.pass_through.typesafe_logging_handler import TypeSafePassthroughLoggingHandler
from litellm.router_strategy.complexity_router.config import DEFAULT_JEV_INSTRUCTIONS as _DEFAULT_JEV_INSTRUCTIONS
@ -33,6 +33,7 @@ if TYPE_CHECKING:
JevProbability: TypeAlias = Annotated[float, Field(ge=0.0, le=1.0)]
ClassifierProvider: TypeAlias = Literal["typesafe", "laya", "bespoke", "databricks"]
DEFAULT_JEV_INSTRUCTIONS: Final = _DEFAULT_JEV_INSTRUCTIONS
_WITHOUT_MODEL: Final = MappingProxyType({"model": True})
class JevChoiceQuestion(LiteLLMBaseModel):
@ -181,7 +182,7 @@ class HttpJevClassifierClient:
)
response: Final = await self._http_client.post( # pyright: ignore[reportUnknownMemberType] # AsyncHTTPHandler has a dynamic post signature
self._request_url(request.model),
json=request.model_dump(mode="json"),
json=request.model_dump(mode="json", exclude=self._excluded_fields(request.model)),
headers=MappingProxyType({**authorization, "Content-Type": "application/json"}), # pyright: ignore[reportArgumentType] # HTTP headers are not mutated by AsyncHTTPHandler
timeout=timeout_s,
)
@ -194,6 +195,9 @@ class HttpJevClassifierClient:
verbose_router_logger.warning("JEV response logging failed (%s)", type(exc).__name__)
return TypeAdapter(JevSystemOneResponse).validate_python(normalized_body)
def _excluded_fields(self, model: str) -> Mapping[str, bool] | None:
return _WITHOUT_MODEL if self._provider == "databricks" and is_ai_decide_model(model) else None
def _request_url(self, model: str) -> str:
if self._provider == "databricks":
return DATABRICKS_DECISIONS_CONFIG.get_complete_url(self._api_base, model)

View file

@ -21177,6 +21177,18 @@
"output_cost_per_token": 0.0,
"source": "https://docs.databricks.com/aws/en/machine-learning/foundation-model-apis/supported-models"
},
"databricks/ai_decide": {
"cache_creation_input_token_cost": 0.0,
"cache_read_input_token_cost": 0.0,
"input_cost_per_token": 0.0,
"litellm_provider": "databricks",
"metadata": {
"notes": "The ai_decide AI Function (Beta), called through POST /api/2.0/ai-functions/ai-decide. Its response carries no token usage, so LiteLLM records no spend for it."
},
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.databricks.com/api/ai-functions/v1/ai-decide"
},
"dataforseo/search": {
"input_cost_per_query": 0.003,
"litellm_provider": "dataforseo",

View file

@ -2,28 +2,34 @@ import asyncio
import json
import threading
import uuid
from collections.abc import Callable, Mapping
from collections.abc import Callable, Mapping, Sequence
from queue import SimpleQueue
from typing import Final
import httpx
import pytest
from integration._support.client import Gateway, Scenario, eventually, object_value, string_value
from integration._support.database import read_rows
from integration._support.openai_wire import answering_model_discovery, chat_reply
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue
from litellm.router_strategy.complexity_router.config import DEFAULT_JEV_INSTRUCTIONS
_MODEL: Final = "ai_decide"
_ROUTE: Final = "/api/2.0/ai-functions/ai-decide"
_ENDPOINT: Final = "databricks-openjev-qwen35-4b"
_SECOND_ENDPOINT: Final = "databricks-openjev-qwen35-4b-canary"
_SECOND_WORKSPACE: Final = "/second-workspace"
_API_KEY: Final = "synthetic-databricks-key"
_BARE_NAME_RULE: Final = (
"must be the bare endpoint name (letters, digits, '-', '_' and '.', with no '/', '?', '#', spaces, or a "
"leading '.'), e.g. databricks-openjev-qwen35-4b"
_MODEL_RULE: Final = (
"Databricks decisions model must be 'ai_decide' (the ai_decide AI Function) or a bare serving endpoint name "
"(letters, digits, '-', '_' and '.', with no '/', '?', '#', spaces, or a leading '.'), e.g. "
"databricks-openjev-qwen35-4b"
)
_REJECTED_AT_LOAD: Final = "The router would drop this deployment at load time, so the write is rejected instead."
_MODEL_REQUIRED: Final = (
"opensource_classifier_config.model is required for provider 'databricks': the serving endpoint name, "
"e.g. databricks-openjev-qwen35-4b"
"opensource_classifier_config.model is required for provider 'databricks': 'ai_decide' or a serving endpoint "
"name, e.g. databricks-openjev-qwen35-4b"
)
_BASE_NEEDS_KEY: Final = (
"opensource_classifier_config.api_base requires opensource_classifier_config.api_key: DATABRICKS_API_KEY or "
@ -32,17 +38,26 @@ _BASE_NEEDS_KEY: Final = (
_BLANK_KEY: Final = (
"opensource_classifier_config.api_key must be non-empty; omit it to use the provider environment key"
)
_CLASSIFIER_ANSWER: Final[dict[str, JsonValue]] = {
"answers": {
"tier": {
"type": "choice",
"choice": "COMPLEX",
"confidence": 0.91,
"probabilities": {"SIMPLE": 0.03, "MEDIUM": 0.06, "COMPLEX": 0.91},
}
},
_ANSWERS: Final[dict[str, JsonValue]] = {
"tier": {
"type": "choice",
"choice": "COMPLEX",
"confidence": 0.91,
"probabilities": {"SIMPLE": 0.03, "MEDIUM": 0.06, "COMPLEX": 0.91},
}
}
_AI_DECIDE_ANSWER: Final[dict[str, JsonValue]] = {"response": {"answers": _ANSWERS}, "metadata": {"version": "1.0"}}
_ENDPOINT_ANSWER: Final[dict[str, JsonValue]] = {
"model": "/mosaicml/local_model",
"answers": _ANSWERS,
"usage": {"input_tokens": 12, "output_tokens": 1},
}
_TIER_CRITERIA: Final[dict[str, JsonValue]] = {
"SIMPLE": "Greetings, chitchat, or short factual lookups with known answers",
"MEDIUM": "Everyday requests needing explanation, light reasoning, or minor technical work",
"COMPLEX": "Non-trivial code, architecture, multi-step work, or specialized domain depth",
"REASONING": "Open-ended analysis, proofs, tradeoffs, or tasks requiring careful thought",
}
_ROWS_QUERY: Final = (
"SELECT request_id, status, metadata->>'internal_call_origin' AS origin "
'FROM "LiteLLM_SpendLogs" WHERE model_group = %s'
@ -50,7 +65,8 @@ _ROWS_QUERY: Final = (
def _classifier(request: Request) -> Reply:
return Reply(body=json.dumps(_CLASSIFIER_ANSWER).encode())
answer: Final = _ENDPOINT_ANSWER if request.target.endswith("/invocations") else _AI_DECIDE_ANSWER
return Reply(body=json.dumps(answer).encode())
def _tier(text: str) -> Callable[[Request], Reply]:
@ -64,7 +80,7 @@ def _tier(text: str) -> Callable[[Request], Reply]:
def _classifier_config(api_base: str, **overrides: JsonValue) -> dict[str, JsonValue]:
return {
"provider": "databricks",
"model": _ENDPOINT,
"model": _MODEL,
"api_base": api_base,
"api_key": _API_KEY,
"timeout_ms": 20000,
@ -133,21 +149,39 @@ async def _burst(base_url: str, key: str, model: str, count: int) -> tuple[httpx
)
def _rows(router: str, *, want: int) -> tuple[dict[str, JsonValue], ...]:
return tuple(eventually(lambda: read_rows(_ROWS_QUERY, (router,)), lambda found: len(found) >= want, seconds=70))
def _request_rows(rows: Sequence[dict[str, JsonValue]]) -> tuple[dict[str, JsonValue], ...]:
return tuple(row for row in rows if row["origin"] != "autorouter_classifier")
def _rows(router: str, *, requests: int) -> tuple[dict[str, JsonValue], ...]:
return tuple(
eventually(
lambda: read_rows(_ROWS_QUERY, (router,)),
lambda found: len(_request_rows(found)) >= requests,
seconds=70,
)
)
def test_validate_accepts_a_databricks_ai_decide_classifier_with_or_without_the_serving_endpoints_suffix(
gateway: Gateway,
) -> None:
for api_base in (
"https://dbc-example.cloud.databricks.com",
"https://dbc-example.cloud.databricks.com/serving-endpoints",
):
config: Final = _router_config(_classifier_config(api_base), "simple", "complex")
assert _validate(gateway, config) == {"valid": True, "error": None}, api_base
def test_validate_accepts_a_databricks_classifier_with_a_bare_endpoint_name_of_any_length(gateway: Gateway) -> None:
config: Final = _router_config(
_classifier_config("https://dbc-example.cloud.databricks.com/serving-endpoints"), "simple", "complex"
)
assert _validate(gateway, config) == {"valid": True, "error": None}
long_name: Final = _router_config(
_classifier_config("https://dbc-example.cloud.databricks.com/serving-endpoints", model="e" * 5000),
"simple",
"complex",
)
assert _validate(gateway, long_name) == {"valid": True, "error": None}
for model in (_ENDPOINT, "e" * 5000):
config: Final = _router_config(
_classifier_config("https://dbc-example.cloud.databricks.com/serving-endpoints", model=model),
"simple",
"complex",
)
assert _validate(gateway, config) == {"valid": True, "error": None}, model[:40]
def test_validate_names_each_databricks_classifier_misconfiguration(gateway: Gateway) -> None:
@ -159,10 +193,10 @@ def test_validate_names_each_databricks_classifier_misconfiguration(gateway: Gat
_without(base, "api_key"),
f"at opensource_classifier_config: Value error, {_BASE_NEEDS_KEY}",
),
("slash in the name", {**base, "model": "a/b"}, f"Databricks serving endpoint name 'a/b' {_BARE_NAME_RULE}"),
("space in the name", {**base, "model": "a b"}, f"Databricks serving endpoint name 'a b' {_BARE_NAME_RULE}"),
("leading dot", {**base, "model": ".hidden"}, f"Databricks serving endpoint name '.hidden' {_BARE_NAME_RULE}"),
("empty name", {**base, "model": ""}, f"Databricks serving endpoint name '' {_BARE_NAME_RULE}"),
("slash in the name", {**base, "model": "a/b"}, _MODEL_RULE),
("space in the name", {**base, "model": "a b"}, _MODEL_RULE),
("leading dot", {**base, "model": ".hidden"}, _MODEL_RULE),
("empty name", {**base, "model": ""}, _MODEL_RULE),
(
"blank api_key",
{**base, "api_key": ""},
@ -185,8 +219,13 @@ def test_validate_names_each_databricks_classifier_misconfiguration(gateway: Gat
assert error.endswith(_REJECTED_AT_LOAD), (label, error)
def test_test_routing_classifies_through_the_databricks_endpoint_without_calling_the_tier_it_picked(
gateway: Gateway,
@pytest.mark.parametrize(
("model", "target", "cost"),
[(_MODEL, _ROUTE, {}), (_ENDPOINT, f"/{_ENDPOINT}/invocations", {"classifier_cost": 0.0})],
ids=["ai_decide", "serving_endpoint"],
)
def test_test_routing_classifies_through_databricks_without_calling_the_tier_it_picked(
gateway: Gateway, model: str, target: str, cost: dict[str, JsonValue]
) -> None:
prompt: Final = f"design a distributed cache {uuid.uuid4().hex}"
with (
@ -197,7 +236,7 @@ def test_test_routing_classifies_through_the_databricks_endpoint_without_calling
):
simple_model: Final = scenario.model(model="openai/simple-tier", api_base=simple.url)
complex_model: Final = scenario.model(model="openai/complex-tier", api_base=complex_tier.url)
config: Final = _router_config(_classifier_config(judge.url), simple_model, complex_model)
config: Final = _router_config(_classifier_config(judge.url, model=model), simple_model, complex_model)
response: Final = gateway.request(
"POST", "/auto_router/test_routing", {"prompt": prompt, "complexity_router_config": config}
)
@ -218,23 +257,27 @@ def test_test_routing_classifies_through_the_databricks_endpoint_without_calling
"tier-probability:MEDIUM=0.060000",
"tier-probability:COMPLEX=0.910000",
],
"classifier_model": f"databricks/{_ENDPOINT}",
"classifier_cost": 0.0,
"classifier_model": f"databricks/{model}",
**cost,
"classifier_probabilities": {"SIMPLE": 0.03, "MEDIUM": 0.06, "COMPLEX": 0.91},
"classifier_confidence": 0.91,
"conversation_continuing": False,
},
}, response.text
(call,) = [request for request in judge.drain() if request.method == "POST"]
assert call.target == f"/{_ENDPOINT}/invocations", call.target
assert call.target == target, call.target
assert call.headers.get("authorization") == f"Bearer {_API_KEY}", call.headers
body: Final = json.loads(call.body)
assert body["model"] == _ENDPOINT, body
assert body["state"] == f"\nClassify this message:\n{prompt}", body
assert json.loads(call.body) == {
"state": f"\nClassify this message:\n{prompt}",
**({} if model == _MODEL else {"model": model}),
"questions": {
"tier": {"type": "choice", "instructions": DEFAULT_JEV_INSTRUCTIONS, "criteria": _TIER_CRITERIA}
},
}, call.body
assert _judge_targets(simple) == () and _judge_targets(complex_tier) == ()
def test_model_new_refuses_a_databricks_classifier_without_an_endpoint_name_and_stores_nothing(
def test_model_new_refuses_a_databricks_classifier_without_a_model_and_stores_nothing(
gateway: Gateway,
) -> None:
name: Final = f"router-{uuid.uuid4().hex[:10]}"
@ -261,7 +304,7 @@ def test_model_new_refuses_a_databricks_classifier_without_an_endpoint_name_and_
assert [entry for entry in map(object_value, listed) if entry["model_name"] == name] == []
def test_model_update_refuses_dropping_the_endpoint_name_and_keeps_the_stored_classifier_serving(
def test_model_update_refuses_dropping_the_model_and_keeps_the_stored_classifier_serving(
gateway: Gateway,
) -> None:
with (
@ -284,15 +327,15 @@ def test_model_update_refuses_dropping_the_endpoint_name_and_keeps_the_stored_cl
assert _MODEL_REQUIRED in string_value(object_value(object_value(response.json())["error"])["message"]), (
response.text
)
assert _stored_classifier(gateway, identity)["model"] == _ENDPOINT
assert _stored_classifier(gateway, identity)["model"] == _MODEL
chat: Final = gateway.request("POST", "/v1/chat/completions", _chat_body(name))
assert chat.status_code == 200, chat.text
assert chat.json()["choices"][0]["message"]["content"] == "complex answer", chat.text
assert chat.headers["x-litellm-complexity-router-cause"] == "jev_classifier", dict(chat.headers)
assert _judge_targets(judge) == (f"/{_ENDPOINT}/invocations",)
assert _judge_targets(judge) == (_ROUTE,)
async def test_renaming_the_endpoint_under_a_burst_moves_the_classifier_and_logs_every_request_once(
async def test_moving_the_workspace_under_a_burst_moves_the_classifier_and_logs_every_request_once(
gateway: Gateway,
) -> None:
base_url: Final = str(gateway.client.base_url)
@ -315,7 +358,9 @@ async def test_renaming_the_endpoint_under_a_burst_moves_the_classifier_and_logs
complex_model: Final = scenario.model(model="openai/complex-tier", api_base=complex_tier.url)
classifier: Final = _classifier_config(judge.url)
name, identity = _create_router(scenario, _router_config(classifier, simple_model, complex_model))
renamed: Final = _router_config({**classifier, "model": _SECOND_ENDPOINT}, simple_model, complex_model)
moved: Final = _router_config(
{**classifier, "api_base": f"{judge.url}{_SECOND_WORKSPACE}"}, simple_model, complex_model
)
async with asyncio.TaskGroup() as requests:
burst: Final = requests.create_task(_burst(base_url, gateway.key, name, 24))
try:
@ -323,15 +368,15 @@ async def test_renaming_the_endpoint_under_a_burst_moves_the_classifier_and_logs
patched: Final = gateway.request(
"PATCH",
f"/model/{identity}/update",
{"litellm_params": {"model": "auto_router/complexity_router", "complexity_router_config": renamed}},
{"litellm_params": {"model": "auto_router/complexity_router", "complexity_router_config": moved}},
)
assert patched.status_code == 200, patched.text
finally:
release.set()
during: Final = await burst
assert [response.status_code for response in during] == [200] * 24, [response.text for response in during]
assert set(_judge_targets(judge)) <= {f"/{_ENDPOINT}/invocations", f"/{_SECOND_ENDPOINT}/invocations"}
assert _stored_classifier(gateway, identity)["model"] == _SECOND_ENDPOINT
assert set(_judge_targets(judge)) <= {_ROUTE, f"{_SECOND_WORKSPACE}{_ROUTE}"}
assert _stored_classifier(gateway, identity)["api_base"] == f"{judge.url}{_SECOND_WORKSPACE}"
convergence_calls: Final[SimpleQueue[str]] = SimpleQueue()
@ -341,16 +386,16 @@ async def test_renaming_the_endpoint_under_a_burst_moves_the_classifier_and_logs
convergence_calls.put(string_value(chat.json()["id"]))
return _judge_targets(judge)
eventually(classifier_target, lambda targets: targets == (f"/{_SECOND_ENDPOINT}/invocations",), seconds=30)
eventually(classifier_target, lambda targets: targets == (f"{_SECOND_WORKSPACE}{_ROUTE}",), seconds=30)
after: Final = await _burst(base_url, gateway.key, name, 6)
assert [response.status_code for response in after] == [200] * 6, [response.text for response in after]
assert set(_judge_targets(judge)) == {f"/{_SECOND_ENDPOINT}/invocations"}
assert set(_judge_targets(judge)) == {f"{_SECOND_WORKSPACE}{_ROUTE}"}
identities: Final = (
*tuple(string_value(response.json()["id"]) for response in (*during, *after)),
*tuple(convergence_calls.get_nowait() for _ in range(convergence_calls.qsize())),
)
assert len(set(identities)) == len(identities), identities
rows: Final = _rows(name, want=len(identities))
request_rows: Final = [row for row in rows if row["origin"] != "autorouter_classifier"]
rows: Final = _rows(name, requests=len(identities))
request_rows: Final = _request_rows(rows)
assert sorted(string_value(row["request_id"]) for row in request_rows) == sorted(identities), request_rows
assert all(row["status"] == "success" for row in rows), rows

View file

@ -190,7 +190,7 @@ def test_evaluation_mode_health_check_of_hosted_vllm_sends_a_choice_probe(gatewa
]
_DATABRICKS_PROBE_REPLY: Final = JsonResponse(
_DATABRICKS_ENDPOINT_PROBE_REPLY: Final = JsonResponse(
content_type="application/json",
body={
"model": "databricks-openjev-qwen35-4b",
@ -204,7 +204,7 @@ def test_evaluation_mode_health_check_of_a_databricks_serving_endpoint_resolves_
gateway: Gateway,
) -> None:
with gateway.scenario() as scenario:
handle: Final = register_scenario(f"health-decisions-{uuid.uuid4().hex[:12]}", _DATABRICKS_PROBE_REPLY)
handle: Final = register_scenario(f"health-decisions-{uuid.uuid4().hex[:12]}", _DATABRICKS_ENDPOINT_PROBE_REPLY)
scenario.cleanups.callback(delete_scenario, handle)
model: Final = scenario.model(
model="databricks/databricks-openjev-qwen35-4b",
@ -228,3 +228,41 @@ def test_evaluation_mode_health_check_of_a_databricks_serving_endpoint_resolves_
},
)
]
_DATABRICKS_AI_DECIDE_PROBE_REPLY: Final = JsonResponse(
content_type="application/json",
body={
"response": {"answers": {"reachable": {"type": "noul", "probability": 1.0}}},
"metadata": {"version": "1.0"},
},
)
def test_evaluation_mode_health_check_of_databricks_ai_decide_resolves_the_mode_from_the_cost_map(
gateway: Gateway,
) -> None:
with gateway.scenario() as scenario:
handle: Final = register_scenario(f"health-decisions-{uuid.uuid4().hex[:12]}", _DATABRICKS_AI_DECIDE_PROBE_REPLY)
scenario.cleanups.callback(delete_scenario, handle)
model: Final = scenario.model(
model="databricks/ai_decide",
api_base=handle.api_base(),
api_key="synthetic-databricks-key",
)
listed: Final = gateway.request("GET", "/v2/model/info", params={"model": model})
assert listed.status_code == 200, listed.text
assert [object_value(object_value(entry)["model_info"])["mode"] for entry in listed.json()["data"]] == [
"evaluation"
], listed.text
report: Final = _health_report(gateway, model)
assert (report["healthy_count"], report["unhealthy_count"]) == (1, 0), report
assert _probes_sent_to(gateway, handle) == [
(
f"/{handle.scenario_id}/api/2.0/ai-functions/ai-decide",
{
"state": os.environ.get("DEFAULT_HEALTH_CHECK_PROMPT", "test from litellm"),
"questions": {"reachable": {"type": "noul", "instructions": "Is the service reachable?"}},
},
)
]

View file

@ -14,6 +14,8 @@ from integration._support.upstream import ScenarioHandle, delete_scenario, regis
from integration.cost_calculation.cost_tracking_case import JsonResponse
from pydantic import JsonValue
_MODEL: Final = "ai_decide"
_ROUTE: Final = "/api/2.0/ai-functions/ai-decide"
_ENDPOINT: Final = "databricks-openjev-qwen35-4b"
_API_KEY: Final = "synthetic-databricks-key"
_ENV_KEY: Final = "synthetic-databricks-env-key"
@ -38,9 +40,18 @@ _ANSWERS: Final[dict[str, JsonValue]] = {
}
_USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 367, "output_tokens": 3}
_ANSWER_BODY: Final[dict[str, JsonValue]] = {"model": _ENDPOINT, "answers": _ANSWERS, "usage": _USAGE}
_BARE_NAME_RULE: Final = (
"must be the bare endpoint name (letters, digits, '-', '_' and '.', with no '/', '?', '#', spaces, or a "
"leading '.'), e.g. databricks-openjev-qwen35-4b"
_METADATA: Final[dict[str, JsonValue]] = {"version": "1.0"}
_AI_DECIDE_BODY: Final[dict[str, JsonValue]] = {"response": {"answers": _ANSWERS}, "metadata": _METADATA}
_SYSTEM_ONE_BODY: Final[dict[str, JsonValue]] = {
"model": None,
"answers": _ANSWERS,
"usage": {"input_tokens": 0, "output_tokens": 0},
"metadata": _METADATA,
}
_MODEL_RULE: Final = (
"Databricks decisions model must be 'ai_decide' (the ai_decide AI Function) or a bare serving endpoint name "
"(letters, digits, '-', '_' and '.', with no '/', '?', '#', spaces, or a leading '.'), e.g. "
"databricks-openjev-qwen35-4b"
)
_MISSING_KEY: Final = "Missing API key for Decisions provider 'databricks'"
_NON_BARE_NAMES: Final[tuple[tuple[str, str], ...]] = (
@ -49,10 +60,17 @@ _NON_BARE_NAMES: Final[tuple[tuple[str, str], ...]] = (
("leading-dot", ".hidden"),
("question-mark", "a?b"),
("hash", "a#b"),
("route-path", "serving-endpoints/ai_decide"),
)
_DATABRICKS_ERRORS: Final[tuple[tuple[int, dict[str, JsonValue]], ...]] = (
(400, {"error_code": "BAD_REQUEST", "message": "Invalid input: questions must not be empty"}),
(404, {"error_code": "RESOURCE_DOES_NOT_EXIST", "message": f"Endpoint with name '{_ENDPOINT}' does not exist."}),
_DATABRICKS_ERRORS: Final[tuple[tuple[str, str, int, dict[str, JsonValue]], ...]] = (
(_MODEL, _ROUTE, 400, {"error_code": "BAD_REQUEST", "message": "Invalid input: questions must not be empty"}),
(_MODEL, _ROUTE, 403, {"error_code": "PERMISSION_DENIED", "message": "ai_decide is not enabled for this workspace"}),
(
_ENDPOINT,
f"/{_ENDPOINT}/invocations",
404,
{"error_code": "RESOURCE_DOES_NOT_EXIST", "message": f"Endpoint with name '{_ENDPOINT}' does not exist."},
),
)
_SPEND_QUERY: Final = (
'SELECT spend, status, call_type, model_group, custom_llm_provider, api_base FROM "LiteLLM_SpendLogs" '
@ -60,6 +78,7 @@ _SPEND_QUERY: Final = (
)
_OWNED_PROXY_BUDGET: Final = int(2 * graceful_stop_seconds() + 120)
_MEGABYTE_NAME: Final = "a" * 1_000_000 + "/"
_TENTH_MEGABYTE_NAME: Final = "a" * 100_000 + "/"
def _number(value: JsonValue) -> float:
@ -92,6 +111,12 @@ def _upstream_calls(gateway: Gateway, handle: ScenarioHandle) -> list[dict[str,
return _calls_to(_observed(gateway), handle)
def _timed_decide(gateway: Gateway, model: str) -> tuple[httpx.Response, float]:
started: Final = time.perf_counter()
response: Final = _decide(gateway, model)
return response, time.perf_counter() - started
def _spend_row(call_id: str) -> dict[str, JsonValue]:
rows: Final = eventually(lambda: read_rows(_SPEND_QUERY, (call_id,)), lambda found: len(found) == 1, seconds=70)
return rows[0]
@ -116,7 +141,7 @@ def _router_config(classifier: Mapping[str, JsonValue], simple: str, complex_: s
def _environment_classifier() -> dict[str, JsonValue]:
return {"provider": "databricks", "model": _ENDPOINT, "timeout_ms": 20000, "circuit_breaker_enabled": False}
return {"provider": "databricks", "model": _MODEL, "timeout_ms": 20000, "circuit_breaker_enabled": False}
def _router(scenario: Scenario, classifier: Mapping[str, JsonValue]) -> str:
@ -155,8 +180,8 @@ def _assert_classified_by_the_complex_tier(response: httpx.Response, router: str
response.headers["x-litellm-model-name"],
response.headers["x-litellm-complexity-router-tier"],
response.headers["x-litellm-complexity-router-cause"],
response.headers["x-litellm-classifier-cost"],
) == (router, "openai/complex-tier", "COMPLEX", "jev_classifier", "0.0"), dict(response.headers)
) == (router, "openai/complex-tier", "COMPLEX", "jev_classifier"), dict(response.headers)
assert "x-litellm-classifier-cost" not in response.headers, dict(response.headers)
def _assert_classifier_call(
@ -165,14 +190,14 @@ def _assert_classifier_call(
assert call["path"] == path, call
assert call["authorization"] == authorization, call
body: Final = object_value(call["body"])
assert body["model"] == _ENDPOINT, body
assert "model" not in body, body
assert prompt_marker in string_value(body["state"]), body
assert sorted(object_value(object_value(body["questions"])["tier"])) == ["criteria", "instructions", "type"], body
def _assert_refused_before_any_upstream_call(gateway: Gateway, model: str, message: str) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _ANSWER_BODY)
handle: Final = _register(scenario, _AI_DECIDE_BODY)
deployment: Final = scenario.model(model=model, api_base=handle.api_base(), api_key=_API_KEY)
response: Final = _decide(gateway, deployment)
assert response.status_code == 400, response.text
@ -186,12 +211,10 @@ def _assert_refused_before_any_upstream_call(gateway: Gateway, model: str, messa
def test_a_deployment_naming_a_non_bare_endpoint_is_refused_before_any_upstream_call(
gateway: Gateway, name: str
) -> None:
_assert_refused_before_any_upstream_call(
gateway, f"databricks/{name}", f"Databricks serving endpoint name {name!r} {_BARE_NAME_RULE}"
)
_assert_refused_before_any_upstream_call(gateway, f"databricks/{name}", _MODEL_RULE)
def test_a_deployment_naming_no_endpoint_is_refused_before_any_upstream_call(gateway: Gateway) -> None:
def test_a_deployment_naming_no_model_is_refused_before_any_upstream_call(gateway: Gateway) -> None:
_assert_refused_before_any_upstream_call(gateway, "databricks/", "A model name is required for the Decisions API")
@ -256,6 +279,81 @@ def test_an_openai_format_request_at_v1_decisions_reaches_the_serving_endpoint_a
)
def test_an_openai_format_request_at_v1_decisions_reaches_the_ai_decide_route_without_a_model_field(
gateway: Gateway,
) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _AI_DECIDE_BODY)
deployment: Final = scenario.model(model=f"databricks/{_MODEL}", api_base=handle.api_base(), api_key=_API_KEY)
response: Final = gateway.request(
"POST",
"/v1/decisions",
{
"model": deployment,
"input": _OPENAI_INPUT,
"questions": [_OPENAI_TIER_QUESTION],
},
)
assert response.status_code == 200, response.text
assert response.json() == {
"model": f"databricks/{_MODEL}",
"answers": [
{
"type": "choice",
"name": "tier",
"choice": "COMPLEX",
"probabilities": [
{"value": "SIMPLE", "probability": 0.03},
{"value": "MEDIUM", "probability": 0.06},
{"value": "COMPLEX", "probability": 0.91},
],
"confidence": 0.91,
}
],
"usage": {
"input_tokens": 0,
"input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0},
"output_tokens": 0,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": 0,
},
}, response.text
assert response.headers["x-litellm-model-name"] == f"databricks/{_MODEL}", dict(response.headers)
(call,) = _upstream_calls(gateway, handle)
assert call["path"] == f"/{handle.scenario_id}{_ROUTE}", call
assert call["authorization"] == f"Bearer {_API_KEY}", call
assert call["body"] == {
"state": _OPENAI_INPUT,
"questions": {"tier": {"type": "choice", "instructions": "Which tier?", "criteria": _TIER_CRITERIA}},
}, call
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (
row["status"],
row["call_type"],
row["custom_llm_provider"],
row["model_group"],
row["api_base"],
_number(row["spend"]),
) == ("success", "adecisions", "databricks", deployment, f"{handle.api_base()}{_ROUTE}", 0.0), row
def test_a_noul_probability_reaches_the_caller_as_the_system_one_noul_answer(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(
scenario, {"response": {"answers": {"defect": {"type": "noul", "probability": 0.8}}}, "metadata": _METADATA}
)
deployment: Final = scenario.model(model=f"databricks/{_MODEL}", api_base=handle.api_base(), api_key=_API_KEY)
response: Final = gateway.request(
"POST",
"/v1/systemone",
{"model": deployment, "state": _STATE, "questions": {"defect": {"type": "noul", "instructions": "Bug?"}}},
)
assert response.status_code == 200, response.text
assert response.json()["answers"] == {"defect": {"type": "noul", "noul": 0.8}}, response.text
(call,) = _upstream_calls(gateway, handle)
assert call["body"] == {"state": _STATE, "questions": {"defect": {"type": "noul", "instructions": "Bug?"}}}
def test_a_wildcard_deployment_sends_each_request_to_the_named_endpoint_and_bills_each_once(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _ANSWER_BODY)
@ -289,12 +387,43 @@ def test_a_wildcard_deployment_sends_each_request_to_the_named_endpoint_and_bill
)
def test_a_wildcard_deployment_sends_each_ai_decide_request_to_the_route_and_logs_each_once(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _AI_DECIDE_BODY)
prefix: Final = _wildcard_deployment(scenario, api_base=handle.api_base(), api_key=_API_KEY)
model: Final = f"{prefix}/{_MODEL}"
responses: Final = tuple(_decide(gateway, model) for _ in range(2))
for response in responses:
assert response.status_code == 200, response.text
assert response.json() == _SYSTEM_ONE_BODY, response.text
assert response.headers["x-litellm-model-group"] == model, dict(response.headers)
assert response.headers["x-litellm-model-name"] == f"databricks/{_MODEL}", dict(response.headers)
call_ids: Final = tuple(response.headers["x-litellm-call-id"] for response in responses)
assert len(set(call_ids)) == 2, call_ids
calls: Final = _upstream_calls(gateway, handle)
assert len(calls) == 2, calls
for call in calls:
assert call["path"] == f"/{handle.scenario_id}{_ROUTE}", call
assert call["authorization"] == f"Bearer {_API_KEY}", call
assert call["body"] == {"state": _STATE, "questions": _QUESTIONS}, call
for call_id in call_ids:
row: Final = _spend_row(call_id)
assert (
row["status"],
row["call_type"],
row["custom_llm_provider"],
row["model_group"],
row["api_base"],
_number(row["spend"]),
) == ("success", "asystemone", "databricks", model, f"{handle.api_base()}{_ROUTE}", 0.0), row
@pytest.mark.parametrize("model", (5, ["a"]), ids=("int", "list"))
def test_a_non_string_model_is_refused_at_the_gateway_without_an_upstream_call(
gateway: Gateway, model: JsonValue
) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _ANSWER_BODY)
handle: Final = _register(scenario, _AI_DECIDE_BODY)
_wildcard_deployment(scenario, api_base=handle.api_base(), api_key=_API_KEY)
response: Final = gateway.request(
"POST", "/v1/systemone", {"model": model, "state": _STATE, "questions": _QUESTIONS}
@ -328,18 +457,20 @@ def test_a_five_kilobyte_endpoint_name_reaches_the_upstream_and_its_not_found_an
), row
@pytest.mark.parametrize("status,body", _DATABRICKS_ERRORS, ids=("400", "404"))
@pytest.mark.parametrize(
"name,path,status,body", _DATABRICKS_ERRORS, ids=("ai_decide-400", "ai_decide-403", "serving_endpoint-404")
)
def test_a_databricks_error_keeps_its_status_and_message_and_bills_nothing(
gateway: Gateway, status: int, body: dict[str, JsonValue]
gateway: Gateway, name: str, path: str, status: int, body: dict[str, JsonValue]
) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, body, status=status)
model: Final = scenario.model(model=f"databricks/{_ENDPOINT}", api_base=handle.api_base(), api_key=_API_KEY)
model: Final = scenario.model(model=f"databricks/{name}", api_base=handle.api_base(), api_key=_API_KEY)
response: Final = _decide(gateway, model)
assert response.status_code == status, response.text
assert string_value(body["message"]) in response.text, response.text
(call,) = _upstream_calls(gateway, handle)
assert call["path"] == f"/{handle.scenario_id}/{_ENDPOINT}/invocations", call
assert call["path"] == f"/{handle.scenario_id}{path}", call
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], row["model_group"], _number(row["spend"])) == (
"failure",
@ -353,8 +484,8 @@ def test_a_deployment_without_a_key_and_no_environment_key_is_refused_before_any
gateway: Gateway,
) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _ANSWER_BODY)
model: Final = scenario.model(model=f"databricks/{_ENDPOINT}", api_base=handle.api_base(), api_key=None)
handle: Final = _register(scenario, _AI_DECIDE_BODY)
model: Final = scenario.model(model=f"databricks/{_MODEL}", api_base=handle.api_base(), api_key=None)
response: Final = _decide(gateway, model)
assert response.status_code == 401, response.text
assert _MISSING_KEY in response.text, response.text
@ -366,9 +497,9 @@ def test_the_environment_key_and_base_serve_a_deployment_a_classifier_and_a_team
gateway: Gateway, tmp_path: Path
) -> None:
with gateway.scenario() as rig_scenario:
handle: Final = _register(rig_scenario, _ANSWER_BODY)
environment_base: Final = f"{handle.api_base()}/serving-endpoints"
environment_path: Final = f"/{handle.scenario_id}/serving-endpoints/{_ENDPOINT}/invocations"
handle: Final = _register(rig_scenario, _AI_DECIDE_BODY)
environment_base: Final = f"{handle.api_base()}/environment-workspace/serving-endpoints"
environment_path: Final = f"/{handle.scenario_id}/environment-workspace{_ROUTE}"
environment: Final = {
"DATABRICKS_API_KEY": _ENV_KEY,
"DATABRICKS_TOKEN": _ENV_TOKEN,
@ -376,14 +507,14 @@ def test_the_environment_key_and_base_serve_a_deployment_a_classifier_and_a_team
}
with owned_proxy_process(gateway, tmp_path, environment) as owned, owned.gateway.scenario() as scenario:
candidate: Final = owned.gateway
with_base: Final = scenario.model(model=f"databricks/{_ENDPOINT}", api_base=handle.api_base(), api_key=None)
bare: Final = scenario.model(model=f"databricks/{_ENDPOINT}", api_base=None, api_key=None)
with_base: Final = scenario.model(model=f"databricks/{_MODEL}", api_base=handle.api_base(), api_key=None)
bare: Final = scenario.model(model=f"databricks/{_MODEL}", api_base=None, api_key=None)
decided: Final = tuple(_decide(candidate, model) for model in (with_base, bare))
for response in decided:
assert response.status_code == 200, response.text
assert response.json() == _ANSWER_BODY, response.text
assert response.json() == _SYSTEM_ONE_BODY, response.text
assert [(call["path"], call["authorization"]) for call in _upstream_calls(candidate, handle)] == [
(f"/{handle.scenario_id}/{_ENDPOINT}/invocations", f"Bearer {_ENV_KEY}"),
(f"/{handle.scenario_id}{_ROUTE}", f"Bearer {_ENV_KEY}"),
(environment_path, f"Bearer {_ENV_KEY}"),
]
for response, model in zip(decided, (with_base, bare)):
@ -391,7 +522,7 @@ def test_the_environment_key_and_base_serve_a_deployment_a_classifier_and_a_team
assert (row["status"], row["model_group"], _number(row["spend"])) == ("success", model, 0.0), row
assert (
_spend_row(decided[1].headers["x-litellm-call-id"])["api_base"]
== f"{environment_base}/{_ENDPOINT}/invocations"
== f"{handle.api_base()}/environment-workspace{_ROUTE}"
)
router: Final = _router(scenario, _environment_classifier())
@ -435,7 +566,7 @@ def test_the_environment_key_and_base_serve_a_deployment_a_classifier_and_a_team
"opensource_classifier_config"
]
)
assert (stored_classifier["provider"], stored_classifier["model"]) == ("databricks", _ENDPOINT), entry
assert (stored_classifier["provider"], stored_classifier["model"]) == ("databricks", _MODEL), entry
assert "api_key" not in stored_classifier and "api_base" not in stored_classifier, entry
member_chat: Final = _chat(candidate, team_router, key=member_key)
_assert_classified_by_the_complex_tier(member_chat, team_router)
@ -453,21 +584,21 @@ def test_the_environment_token_serves_a_deployment_and_a_classifier_when_no_envi
gateway: Gateway, tmp_path: Path
) -> None:
with gateway.scenario() as rig_scenario:
handle: Final = _register(rig_scenario, _ANSWER_BODY)
environment_path: Final = f"/{handle.scenario_id}/serving-endpoints/{_ENDPOINT}/invocations"
handle: Final = _register(rig_scenario, _AI_DECIDE_BODY)
environment_path: Final = f"/{handle.scenario_id}/environment-workspace{_ROUTE}"
environment: Final = {
"DATABRICKS_TOKEN": _ENV_TOKEN,
"DATABRICKS_API_BASE": f"{handle.api_base()}/serving-endpoints",
"DATABRICKS_API_BASE": f"{handle.api_base()}/environment-workspace/serving-endpoints",
}
with (
owned_proxy_process(gateway, tmp_path, environment, remove_environment=("DATABRICKS_API_KEY",)) as owned,
owned.gateway.scenario() as scenario,
):
candidate: Final = owned.gateway
bare: Final = scenario.model(model=f"databricks/{_ENDPOINT}", api_base=None, api_key=None)
bare: Final = scenario.model(model=f"databricks/{_MODEL}", api_base=None, api_key=None)
decided: Final = _decide(candidate, bare)
assert decided.status_code == 200, decided.text
assert decided.json() == _ANSWER_BODY, decided.text
assert decided.json() == _SYSTEM_ONE_BODY, decided.text
(decision_call,) = _upstream_calls(candidate, handle)
assert (decision_call["path"], decision_call["authorization"]) == (environment_path, f"Bearer {_ENV_TOKEN}")
assert _spend_row(decided.headers["x-litellm-call-id"])["status"] == "success"
@ -483,15 +614,16 @@ def test_the_environment_token_serves_a_deployment_and_a_classifier_when_no_envi
)
def test_a_megabyte_endpoint_name_is_refused_in_linear_time_while_liveliness_stays_fast(gateway: Gateway) -> None:
def test_a_megabyte_model_name_is_refused_in_linear_time_while_liveliness_stays_up(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
handle: Final = _register(scenario, _ANSWER_BODY)
handle: Final = _register(scenario, _AI_DECIDE_BODY)
prefix: Final = _wildcard_deployment(scenario, api_base=handle.api_base(), api_key=_API_KEY)
model: Final = f"{prefix}/{_MEGABYTE_NAME}"
anonymous: Final = gateway.client.post(
"/v1/systemone", json={"model": model, "state": _STATE, "questions": _QUESTIONS}
)
assert anonymous.status_code == 401, anonymous.text[:300]
tenth_response, tenth_elapsed = _timed_decide(gateway, f"{prefix}/{_TENTH_MEGABYTE_NAME}")
liveliness: Final[list[tuple[int, float]]] = []
stop: Final = threading.Event()
@ -503,16 +635,15 @@ def test_a_megabyte_endpoint_name_is_refused_in_linear_time_while_liveliness_sta
poller: Final = threading.Thread(target=poll)
poller.start()
started: Final = time.perf_counter()
response: Final = _decide(gateway, model)
elapsed: Final = time.perf_counter() - started
response, elapsed = _timed_decide(gateway, model)
stop.set()
poller.join()
assert elapsed < 3, elapsed
assert tenth_response.status_code == 400, tenth_response.text[:300]
assert elapsed < 25 * tenth_elapsed, (elapsed, tenth_elapsed)
assert liveliness and all(status == 200 for status, _ in liveliness), liveliness
assert max(seconds for _, seconds in liveliness) < 1, liveliness
assert max(seconds for _, seconds in liveliness) < max(1.0, elapsed / 2), (elapsed, liveliness)
assert response.status_code == 400, response.text[:300]
assert _BARE_NAME_RULE in response.text, response.text[-400:]
assert _MODEL_RULE in response.text, response.text[:400]
assert _upstream_calls(gateway, handle) == []
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], _number(row["spend"])) == ("failure", "asystemone", 0.0), row

View file

@ -88,6 +88,17 @@ _USAGE: Final[dict[str, JsonValue]] = {
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": _INPUT_TOKENS + _OUTPUT_TOKENS,
}
_NO_USAGE: Final[dict[str, JsonValue]] = {
"input_tokens": 0,
"input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0},
"output_tokens": 0,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": 0,
}
_AI_DECIDE_ANSWERS: Final[dict[str, JsonValue]] = {
**_SYSTEM_ONE_ANSWERS,
"defect": {"type": "noul", "probability": 0.93},
}
_SDK_QUESTIONS: Final = TypeAdapter(list[dict[str, object]]).validate_python(_QUESTIONS)
_SPEND_QUERY: Final = (
"SELECT spend, status, call_type, model_group, custom_llm_provider, api_base, prompt_tokens, completion_tokens, "
@ -105,11 +116,14 @@ class _Provider:
wraps_result: bool
cost_map_key: str | None
speaks_openai: bool = False
ai_decide: bool = False
provider_reported_cost: float | None = None
def upstream_body(self) -> dict[str, JsonValue]:
if self.speaks_openai:
return {"model": self.body_model, "input": _INPUT, "questions": _QUESTIONS}
if self.ai_decide:
return {"state": _INPUT, "questions": _SYSTEM_ONE_QUESTIONS}
return {"model": self.body_model, "state": _INPUT, "questions": _SYSTEM_ONE_QUESTIONS}
def _with_reported_cost(self, usage: dict[str, JsonValue]) -> dict[str, JsonValue]:
@ -120,6 +134,8 @@ class _Provider:
def upstream_reply(self) -> dict[str, JsonValue]:
if self.speaks_openai:
return {"model": self.body_model, "answers": _ANSWERS, "usage": self._with_reported_cost(_USAGE)}
if self.ai_decide:
return {"response": {"answers": _AI_DECIDE_ANSWERS}, "metadata": {"version": "1.0"}}
answer: Final[dict[str, JsonValue]] = {
"model": self.body_model,
"answers": _SYSTEM_ONE_ANSWERS,
@ -128,8 +144,13 @@ class _Provider:
return {"result": answer, "success": True} if self.wraps_result else answer
def litellm_response(self) -> dict[str, JsonValue]:
if self.ai_decide:
return {"model": self.model, "answers": _ANSWERS, "usage": _NO_USAGE}
return {"model": self.body_model, "answers": _ANSWERS, "usage": _USAGE}
def billed_tokens(self) -> tuple[int, int]:
return (0, 0) if self.ai_decide else (_INPUT_TOKENS, _OUTPUT_TOKENS)
_PROVIDERS: Final = (
_Provider(
@ -165,6 +186,16 @@ _PROVIDERS: Final = (
"cloudflare/@cf/cloudflare/clef",
),
_Provider("hosted_vllm", "hosted_vllm/Qwen/Qwen3-0.6B", "/v1/systemone", "Qwen/Qwen3-0.6B", None, False, None),
_Provider(
"databricks",
"databricks/ai_decide",
"/api/2.0/ai-functions/ai-decide",
"ai_decide",
_API_KEY,
False,
None,
ai_decide=True,
),
_Provider(
"databricks",
"databricks/databricks-openjev-qwen35-4b",
@ -234,7 +265,7 @@ def _response_cost(response: httpx.Response) -> float:
def _provider_id(provider: _Provider) -> str:
return provider.name
return provider.model
def _number(value: JsonValue) -> float:
@ -358,8 +389,7 @@ def test_each_provider_gets_its_own_path_key_and_body_and_is_billed(gateway: Gat
provider.name,
model,
f"{handle.api_base()}{provider.path}",
_INPUT_TOKENS,
_OUTPUT_TOKENS,
*provider.billed_tokens(),
)
assert math.isclose(_number(row["spend"]), expected_spend, rel_tol=1e-9), row

View file

@ -31,9 +31,10 @@ from pydantic import JsonValue, TypeAdapter
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_if_encrypted_with
from litellm.router_strategy.complexity_router.config import DEFAULT_JEV_INSTRUCTIONS
_ENDPOINT: Final = "databricks-openjev-qwen35-4b"
_MODEL: Final = "ai_decide"
_ROUTE: Final = "/api/2.0/ai-functions/ai-decide"
_API_KEY: Final = "synthetic-databricks-key"
_CLASSIFIER_MODEL: Final = f"databricks/{_ENDPOINT}"
_CLASSIFIER_MODEL: Final = f"databricks/{_MODEL}"
_TYPESAFE_MODEL: Final = "jev-1.13.0"
_TYPESAFE_KEY: Final = "synthetic-typesafe-key"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
@ -48,10 +49,12 @@ _TIER_CRITERIA: Final[dict[str, JsonValue]] = {
}
_PROBABILITIES: Final[dict[str, JsonValue]] = {"SIMPLE": 0.03, "MEDIUM": 0.06, "COMPLEX": 0.91}
_USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 12, "output_tokens": 1}
_ANSWER: Final[dict[str, JsonValue]] = {
"answers": {"tier": {"type": "choice", "choice": "COMPLEX", "confidence": 0.91, "probabilities": _PROBABILITIES}},
"usage": _USAGE,
_ANSWERS: Final[dict[str, JsonValue]] = {
"tier": {"type": "choice", "choice": "COMPLEX", "confidence": 0.91, "probabilities": _PROBABILITIES}
}
_METADATA: Final[dict[str, JsonValue]] = {"version": "1.0"}
_ANSWER: Final[dict[str, JsonValue]] = {"response": {"answers": _ANSWERS}, "metadata": _METADATA}
_TYPESAFE_ANSWER: Final[dict[str, JsonValue]] = {"model": _TYPESAFE_MODEL, "answers": _ANSWERS, "usage": _USAGE}
_ROWS_QUERY: Final = (
"SELECT request_id, status, model, call_type, cache_hit, custom_llm_provider, spend, "
"metadata->>'internal_call_origin' AS origin "
@ -155,7 +158,7 @@ def _wires(classifier: Callable[[Request], Reply] | None = None) -> Iterator[tup
def _classifier_config(judge_url: str, **overrides: JsonValue) -> dict[str, JsonValue]:
return {
"provider": "databricks",
"model": _ENDPOINT,
"model": _MODEL,
"api_base": judge_url,
"api_key": _API_KEY,
"timeout_ms": 20000,
@ -209,25 +212,25 @@ def _state(prompt: str, system: str | None) -> str:
def _assert_classifier_call(
call: Request, *, prompt: str, system: str | None = None, endpoint: str = _ENDPOINT, key: str = _API_KEY
call: Request, *, prompt: str, system: str | None = None, key: str = _API_KEY, endpoint: str | None = None
) -> None:
assert call.target == f"/{endpoint}/invocations", call.target
assert call.target == (_ROUTE if endpoint is None else f"/{endpoint}/invocations"), call.target
assert call.headers.get("authorization") == f"Bearer {key}", call.headers
assert json.loads(call.body) == {
"state": _state(prompt, system),
"model": endpoint,
**({} if endpoint is None else {"model": endpoint}),
"questions": {"tier": {"type": "choice", "instructions": DEFAULT_JEV_INSTRUCTIONS, "criteria": _TIER_CRITERIA}},
}, call.body
def _assert_classified(headers: Mapping[str, str], *, router: str, cost: str | None = "0.0") -> None:
def _assert_classified(headers: Mapping[str, str], *, router: str) -> None:
assert (
headers.get("x-litellm-model-group"),
headers.get("x-litellm-model-name"),
headers.get("x-litellm-complexity-router-tier"),
headers.get("x-litellm-complexity-router-cause"),
headers.get("x-litellm-classifier-cost"),
) == (router, "openai/complex-tier", "COMPLEX", "jev_classifier", cost), dict(headers)
) == (router, "openai/complex-tier", "COMPLEX", "jev_classifier", None), dict(headers)
def _assert_fell_back(headers: Mapping[str, str], *, router: str) -> None:
@ -622,10 +625,19 @@ def test_messages_stream_over_httpx_is_classified_and_logged_under_the_message_s
_assert_logged_once(_rows(routed.name, want=2), route="/v1/messages", request_id=identity)
def test_a_non_catalog_endpoint_name_is_classified_without_a_cost_header_and_logged_at_zero(gateway: Gateway) -> None:
def test_a_serving_endpoint_name_is_classified_through_its_invocations_route_and_logged_at_zero(
gateway: Gateway,
) -> None:
prompt: Final = _prompt()
endpoint: Final = "my-jev-endpoint"
with _wires() as (judge, simple, complex_), gateway.scenario() as scenario:
with (
_wires(_answering({"model": "/mosaicml/local_model", "answers": _ANSWERS, "usage": _USAGE})) as (
judge,
simple,
complex_,
),
gateway.scenario() as scenario,
):
routed: Final = _deploy(scenario, judge, simple, complex_, _classifier_config(judge.url, model=endpoint))
response: Final = gateway.request(
"POST", "/v1/chat/completions", _body("/v1/chat/completions", routed.name, prompt, stream=False)
@ -633,7 +645,7 @@ def test_a_non_catalog_endpoint_name_is_classified_without_a_cost_header_and_log
assert response.status_code == 200, response.text
identity, text = _identity_and_text("/v1/chat/completions", _JSON_OBJECT.validate_json(response.content))
assert text == _COMPLEX_TEXT, response.text
_assert_classified(response.headers, router=routed.name, cost=None)
_assert_classified(response.headers, router=routed.name)
(classifier_call,) = _posts(judge)
_assert_classifier_call(classifier_call, prompt=prompt, endpoint=endpoint)
assert len(_posts(complex_)) == 1 and _posts(simple) == ()
@ -686,7 +698,7 @@ _FAILURES: Final = (
_Failure("401", _failing(401), {}, "classifier_error", "MaskedHTTPStatusError", 0),
_Failure(
"malformed",
_answering({"answers": {"tier": {"type": "choice", "choice": "COMPLEX", "confidence": 0.91}}, "usage": _USAGE}),
_answering({"response": {"answers": {"tier": {"type": "choice", "choice": "COMPLEX", "confidence": 0.91}}}}),
{},
"invalid_response",
"ValidationError",
@ -696,15 +708,17 @@ _FAILURES: Final = (
"unknown-tier",
_answering(
{
"answers": {
"tier": {
"type": "choice",
"choice": "GALAXY",
"confidence": 0.91,
"probabilities": {"GALAXY": 0.91},
"response": {
"answers": {
"tier": {
"type": "choice",
"choice": "GALAXY",
"confidence": 0.91,
"probabilities": {"GALAXY": 0.91},
}
}
},
"usage": _USAGE,
"metadata": _METADATA,
}
),
{},
@ -793,7 +807,7 @@ def test_the_default_jev_provider_still_posts_to_systemone_and_bills_from_the_co
prompt: Final = _prompt()
expected_cost: Final = _typesafe_cost()
with (
_wires(_answering({**_ANSWER, "model": _TYPESAFE_MODEL})) as (judge, simple, complex_),
_wires(_answering(_TYPESAFE_ANSWER)) as (judge, simple, complex_),
gateway.scenario() as scenario,
):
classifier: Final[dict[str, JsonValue]] = {
@ -848,7 +862,7 @@ async def test_a_classifier_outage_mid_traffic_falls_back_and_recovers_with_ever
before_prompts: Final = tuple(_prompt() for _ in calls)
before: Final = await _burst(base_url, gateway.key, routed.name, calls, before_prompts)
assert [item.cause for item in before] == ["jev_classifier"] * 12, before
assert [item.classifier_cost for item in before] == ["0.0"] * 12, before
assert [item.classifier_cost for item in before] == [None] * 12, before
assert [item.text for item in before] == [_COMPLEX_TEXT] * 12, before
assert sorted(json.loads(call.body)["state"] for call in _posts(judge)) == sorted(
_state(prompt, None) for prompt in before_prompts

View file

@ -78,7 +78,7 @@ _STRANDS_RESPONSE: Final[Mapping[str, object]] = {
"latency_ms": 3722.17,
}
_DATABRICKS_ENDPOINT: Final = "databricks-openjev-qwen35-4b"
_DATABRICKS_RESPONSE: Final[Mapping[str, object]] = {
_DATABRICKS_ENDPOINT_RESPONSE: Final[Mapping[str, object]] = {
"model": "/mosaicml/local_model",
"answers": {
"is_defect": {"type": "noul", "noul": 0.9},
@ -91,6 +91,21 @@ _DATABRICKS_RESPONSE: Final[Mapping[str, object]] = {
},
"usage": {"input_tokens": 214, "output_tokens": 0},
}
_DATABRICKS_ROUTE: Final = "https://workspace.example/api/2.0/ai-functions/ai-decide"
_DATABRICKS_RESPONSE: Final[Mapping[str, object]] = {
"response": {
"answers": {
"is_defect": {"type": "noul", "probability": 0.9},
"sentiment": {
"type": "choice",
"choice": "positive",
"confidence": 0.8,
"probabilities": {"positive": 0.8, "negative": 0.2},
},
}
},
"metadata": {"version": "1.0"},
}
_PROVIDERS: Final[tuple[tuple[str, str, str, str], ...]] = (
(
"perplexity",
@ -1093,7 +1108,7 @@ async def test_databricks_requires_api_base_before_http(
with pytest.raises(litellm.BadRequestError, match="DATABRICKS_API_BASE"):
await litellm.asystemone(
model=f"databricks/{_DATABRICKS_ENDPOINT}",
model="databricks/ai_decide",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
)
@ -1113,7 +1128,7 @@ async def test_databricks_rejects_a_serving_endpoint_name_that_would_rewrite_the
monkeypatch.setenv("DATABRICKS_API_BASE", "https://workspace.example/serving-endpoints")
monkeypatch.setenv("DATABRICKS_API_KEY", "dapi-key")
with pytest.raises(litellm.BadRequestError, match="bare endpoint name"):
with pytest.raises(litellm.BadRequestError, match="bare serving endpoint name"):
await litellm.asystemone(
model=f"databricks/{endpoint_name}",
state="review",
@ -1128,13 +1143,13 @@ async def test_databricks_requires_a_key_before_http(
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
) -> None:
monkeypatch.setenv("DATABRICKS_API_BASE", "https://workspace.example/serving-endpoints")
monkeypatch.setenv("DATABRICKS_API_BASE", "https://workspace.example")
monkeypatch.delenv("DATABRICKS_API_KEY", raising=False)
monkeypatch.delenv("DATABRICKS_TOKEN", raising=False)
with pytest.raises(litellm.AuthenticationError, match="Missing API key for Decisions provider 'databricks'"):
await litellm.asystemone(
model=f"databricks/{_DATABRICKS_ENDPOINT}",
model="databricks/ai_decide",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
)
@ -1156,7 +1171,7 @@ async def test_databricks_posts_the_jev_body_to_the_serving_endpoint_invocations
monkeypatch.setenv("DATABRICKS_TOKEN", "dapi-token")
route: Final = respx_mock.post(
f"https://workspace.example/serving-endpoints/{_DATABRICKS_ENDPOINT}/invocations"
).respond(json=_DATABRICKS_RESPONSE)
).respond(json=_DATABRICKS_ENDPOINT_RESPONSE)
questions: Final = {
"is_defect": {"type": "noul", "instructions": "Is this a defect?"},
"sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}},
@ -1183,28 +1198,68 @@ async def test_databricks_posts_the_jev_body_to_the_serving_endpoint_invocations
assert response.hidden_params["model"] == f"databricks/{_DATABRICKS_ENDPOINT}"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"api_base",
[
"https://workspace.example",
"https://workspace.example/",
"https://workspace.example/serving-endpoints",
"https://workspace.example/serving-endpoints/",
],
)
async def test_databricks_posts_the_systemone_body_without_model_to_the_ai_decide_route(
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
api_base: str,
) -> None:
monkeypatch.setenv("DATABRICKS_API_BASE", api_base)
monkeypatch.delenv("DATABRICKS_API_KEY", raising=False)
monkeypatch.setenv("DATABRICKS_TOKEN", "dapi-token")
route: Final = respx_mock.post(_DATABRICKS_ROUTE).respond(json=_DATABRICKS_RESPONSE)
questions: Final = {
"is_defect": {"type": "noul", "instructions": "Is this a defect?"},
"sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}},
}
response: Final = await litellm.adecisions(
model="databricks/ai_decide",
state={"ticket": "export hangs"},
questions=questions,
)
assert route.called
sent: Final = respx_mock.calls[0].request
assert sent.headers["authorization"] == "Bearer dapi-token"
assert json.loads(sent.content) == {"state": {"ticket": "export hangs"}, "questions": questions}
assert response.model is None
assert response.answers["is_defect"] == NoulAnswer(type="noul", noul=0.9)
assert isinstance(response.answers["sentiment"], ChoiceAnswer)
assert response.model_extra == {"metadata": {"version": "1.0"}}
assert response.hidden_params["custom_llm_provider"] == "databricks"
assert response.hidden_params["model"] == "databricks/ai_decide"
@pytest.mark.asyncio
async def test_databricks_router_deployment_sends_its_own_key_to_its_own_base(
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
) -> None:
monkeypatch.setenv("DATABRICKS_API_BASE", "https://other-workspace.example/serving-endpoints")
monkeypatch.setenv("DATABRICKS_API_BASE", "https://other-workspace.example")
monkeypatch.setenv("DATABRICKS_API_KEY", "env-key-stays-home")
router: Final = litellm.Router(
model_list=[
{
"model_name": "decider",
"litellm_params": {
"model": f"databricks/{_DATABRICKS_ENDPOINT}",
"api_base": "https://workspace.example/serving-endpoints",
"model": "databricks/ai_decide",
"api_base": "https://workspace.example",
"api_key": "deployment-key",
},
}
]
)
route: Final = respx_mock.post(
f"https://workspace.example/serving-endpoints/{_DATABRICKS_ENDPOINT}/invocations"
).respond(json=_DATABRICKS_RESPONSE)
route: Final = respx_mock.post(_DATABRICKS_ROUTE).respond(json=_DATABRICKS_RESPONSE)
response: Final = await router.asystemone(
model="decider",

View file

@ -3,27 +3,84 @@ from typing import Final
import pytest
from litellm.llms.databricks.decisions.transformation import DatabricksDecisionsConfig
from litellm.types.decisions import (
DecisionsIRPredicateAnswer,
DecisionsIRPredicateQuestion,
DecisionsIRRequest,
DecisionsIRState,
)
_BASE: Final = "https://workspace.example/serving-endpoints"
_HOST: Final = "https://workspace.example"
_BASE: Final = f"{_HOST}/serving-endpoints"
_AI_DECIDE_ROUTE: Final = f"{_HOST}/api/2.0/ai-functions/ai-decide"
_CONFIG: Final = DatabricksDecisionsConfig()
_REQUEST: Final = DecisionsIRRequest(
input=DecisionsIRState(state="export hangs"),
questions=(DecisionsIRPredicateQuestion(name="is_defect", instructions="Is this a defect?", criteria=None),),
)
_QUESTIONS: Final = {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}
@pytest.mark.parametrize("api_base", [_BASE, f"{_BASE}/"])
def test_the_complete_url_is_the_serving_endpoint_invocations_route(api_base: str) -> None:
def test_a_serving_endpoint_name_is_called_on_its_invocations_route(api_base: str) -> None:
assert _CONFIG.get_complete_url(api_base, "openjev") == f"{_BASE}/openjev/invocations"
@pytest.mark.parametrize("api_base", [_HOST, f"{_HOST}/", _BASE, f"{_BASE}/"])
def test_ai_decide_is_called_on_the_workspace_ai_function_route(api_base: str) -> None:
assert _CONFIG.get_complete_url(api_base, "ai_decide") == _AI_DECIDE_ROUTE
@pytest.mark.parametrize(
"name",
["", "serving-endpoints/openjev", "openjev?x=1", "openjev#frag", ".", "..", ".openjev", "open jev", "openjev%2Fx"],
)
def test_a_name_that_would_rewrite_the_url_is_rejected_before_any_request(name: str) -> None:
with pytest.raises(ValueError, match="bare endpoint name"):
with pytest.raises(ValueError, match="bare serving endpoint name"):
_ = _CONFIG.get_complete_url(_BASE, name)
with pytest.raises(ValueError, match="bare endpoint name"):
with pytest.raises(ValueError, match="bare serving endpoint name"):
_ = _CONFIG.canonical_model(name)
def test_the_rejection_does_not_echo_the_rejected_name() -> None:
name: Final = "openjev?token=" + "x" * 4096
with pytest.raises(ValueError, match="bare serving endpoint name") as raised:
_ = _CONFIG.canonical_model(name)
assert "openjev?token=" not in str(raised.value)
def test_a_serving_endpoint_request_keeps_the_model_the_endpoint_expects() -> None:
body: Final = _CONFIG.transform_decisions_request("openjev", _REQUEST, "databricks")
assert body == {"model": "openjev", "state": "export hangs", "questions": _QUESTIONS}
def test_the_ai_decide_request_omits_model_because_the_route_rejects_unexpected_parameters() -> None:
body: Final = _CONFIG.transform_decisions_request("ai_decide", _REQUEST, "databricks")
assert body == {"state": "export hangs", "questions": _QUESTIONS}
def test_the_ai_decide_envelope_is_unwrapped_and_noul_probability_becomes_the_noul_answer() -> None:
payload: Final = {
"response": {"answers": {"is_defect": {"type": "noul", "probability": 0.9}}},
"metadata": {"version": "1.0"},
}
parsed: Final = _CONFIG.parse_response(payload, _REQUEST)
assert parsed.answers == (DecisionsIRPredicateAnswer(probability=0.9),)
assert parsed.model is None
assert dict(parsed.extra) == {"metadata": {"version": "1.0"}}
def test_a_serving_endpoint_answer_is_parsed_as_sent() -> None:
payload: Final = {
"model": "openjev",
"answers": {"is_defect": {"type": "noul", "noul": 0.25}},
"usage": {"input_tokens": 12, "output_tokens": 0},
}
parsed: Final = _CONFIG.parse_response(payload, _REQUEST)
assert parsed.answers == (DecisionsIRPredicateAnswer(probability=0.25),)
assert parsed.model == "openjev"
def test_connection_prefers_the_configured_values_over_the_environment(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("DATABRICKS_API_BASE", "https://env.example/serving-endpoints")
monkeypatch.setenv("DATABRICKS_API_KEY", "dapi-env")
@ -54,3 +111,10 @@ def test_classifier_response_accounts_the_serving_endpoint_instead_of_the_contai
assert normalized["model"] == "openjev"
assert normalized["usage"] == body["usage"]
assert body["model"] == "/mosaicml/local_model"
def test_classifier_response_unwraps_the_ai_decide_envelope_and_accounts_the_configured_model() -> None:
choice: Final = {"type": "choice", "choice": "simple", "confidence": 0.9, "probabilities": {"simple": 0.9}}
body: Final = {"response": {"answers": {"tier": choice}}, "metadata": {"version": "1.0"}}
normalized: Final = _CONFIG.classifier_response(body, "ai_decide")
assert dict(normalized) == {"answers": {"tier": choice}, "metadata": {"version": "1.0"}, "model": "ai_decide"}

View file

@ -304,6 +304,53 @@ def test_proxy_systemone_dispatches_databricks_serving_endpoint(
assert upstream.calls[0].request.headers["authorization"] == "Bearer dapi-deployment"
def test_proxy_systemone_dispatches_databricks_ai_decide(
client: TestClient,
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
) -> None:
monkeypatch.delenv("DATABRICKS_API_BASE", raising=False)
monkeypatch.delenv("DATABRICKS_API_KEY", raising=False)
monkeypatch.delenv("DATABRICKS_TOKEN", raising=False)
router: Final = litellm.Router(
model_list=[
{
"model_name": "databricks-decider",
"litellm_params": {
"model": "databricks/ai_decide",
"api_base": "https://workspace.example/serving-endpoints",
"api_key": "dapi-deployment",
},
}
]
)
monkeypatch.setattr(litellm.proxy.proxy_server, "llm_router", router)
upstream: Final = respx_mock.post("https://workspace.example/api/2.0/ai-functions/ai-decide").respond(
json={
"response": {"answers": {"is_defect": {"type": "noul", "probability": 0.9}}},
"metadata": {"version": "1.0"},
}
)
response: Final = client.post(
"/v1/systemone",
json={
"model": "databricks-decider",
"state": {"source": "proxy-test"},
"questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
},
)
assert response.status_code == 200, response.text
assert response.json()["answers"] == _RESPONSE["answers"]
assert upstream.called
assert json.loads(upstream.calls[0].request.content) == {
"state": {"source": "proxy-test"},
"questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
}
assert upstream.calls[0].request.headers["authorization"] == "Bearer dapi-deployment"
def test_proxy_decisions_without_model_uses_the_proxy_default_model(
client: TestClient,
monkeypatch: pytest.MonkeyPatch,

View file

@ -146,8 +146,8 @@ def test_tier_config_is_normalized_and_unknown_router_extras_are_rejected() -> N
({"provider": "laya", "model": "english", "api_key": "sk-member"}, "api_key"),
({"provider": "bespoke", "model": "nimble-latest", "api_base": "https://collector.invalid"}, "api_base"),
({"provider": "bespoke", "model": "nimble-latest", "api_key": "sk-member"}, "api_key"),
({"provider": "databricks", "model": "my-openjev", "api_base": "https://collector.invalid"}, "opensource_classifier_config"),
({"provider": "databricks", "model": "my-openjev", "api_key": "dapi-member"}, "api_key"),
({"provider": "databricks", "model": "ai_decide", "api_base": "https://collector.invalid"}, "opensource_classifier_config"),
({"provider": "databricks", "model": "ai_decide", "api_key": "dapi-member"}, "api_key"),
],
)
@pytest.mark.parametrize("legacy", [False, True])
@ -168,7 +168,13 @@ def test_members_cannot_move_the_jev_classifier_off_the_proxys_typesafe_account(
@pytest.mark.parametrize(
("provider", "model"),
[("typesafe", "jev-preview"), ("laya", "english"), ("bespoke", "nimble-latest"), ("databricks", "my-openjev")],
[
("typesafe", "jev-preview"),
("laya", "english"),
("bespoke", "nimble-latest"),
("databricks", "ai_decide"),
("databricks", "my-openjev"),
],
)
@pytest.mark.parametrize("legacy", [False, True])
def test_members_can_still_tune_the_jev_classifier(provider: str, model: str, legacy: bool) -> None:

View file

@ -458,6 +458,7 @@ def test_jev_config_requires_classifier_config() -> None:
("laya", "english", "laya"),
("bespoke", "nimble-latest", "bespoke"),
("databricks", "databricks-openjev-qwen35-4b", "databricks"),
("databricks", "ai_decide", "databricks"),
],
)
def test_classifier_aliases_load_and_serialize_one_canonical_config(
@ -540,14 +541,18 @@ async def test_oss_routes_with_its_own_credentials_and_accounts_the_checkpoint(
assert recorder.calls[0]["response_cost"] == pytest.approx(0.31)
def test_databricks_requires_the_bare_serving_endpoint_name_as_the_model() -> None:
@pytest.mark.parametrize("model", ["ai_decide", "my-openjev", "databricks-openjev-qwen35-4b"])
def test_databricks_accepts_ai_decide_or_a_bare_serving_endpoint_name(model: str) -> None:
assert JevClassifierConfig.model_validate({"provider": "databricks", "model": model}).model == model
def test_databricks_requires_a_model_that_cannot_rewrite_the_url() -> None:
with pytest.raises(ValueError, match="model is required for provider 'databricks'"):
JevClassifierConfig.model_validate({"provider": "databricks"})
with pytest.raises(ValueError, match="bare endpoint name"):
with pytest.raises(ValueError, match="bare serving endpoint name"):
JevClassifierConfig.model_validate({"provider": "databricks", "model": "serving-endpoints/my-openjev"})
with pytest.raises(ValueError, match="bare endpoint name"):
with pytest.raises(ValueError, match="bare serving endpoint name"):
JevClassifierConfig.model_validate({"provider": "databricks", "model": ".."})
assert JevClassifierConfig.model_validate({"provider": "databricks", "model": "my-openjev"}).model == "my-openjev"
@pytest.mark.asyncio
@ -608,6 +613,53 @@ async def test_databricks_routes_through_the_serving_endpoint_and_accounts_the_e
assert recorder.calls[0]["response_cost"] == pytest.approx(0.31)
@pytest.mark.asyncio
@pytest.mark.parametrize("configured_key", [False, True])
@pytest.mark.parametrize("legacy", [False, True])
@pytest.mark.parametrize("api_base", ["https://workspace.test", "https://workspace.test/serving-endpoints/"])
async def test_databricks_routes_through_ai_decide_without_a_model_field_and_accounts_ai_decide(
monkeypatch: pytest.MonkeyPatch, configured_key: bool, legacy: bool, api_base: str
) -> None:
monkeypatch.setenv("TYPESAFE_API_KEY", "never-send-typesafe-key")
monkeypatch.setenv("DATABRICKS_API_BASE", api_base)
monkeypatch.delenv("DATABRICKS_API_KEY", raising=False)
monkeypatch.setenv("DATABRICKS_TOKEN", "dapi-env-token")
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
recorder: Final = _UsageRecorder("databricks/ai_decide")
monkeypatch.setattr(litellm, "_async_success_callback", [recorder])
router: Final = ComplexityRouter(
"databricks-route",
litellm.Router(model_list=[]),
{
"classifier_type": "jev" if legacy else "oss_classifier",
"jev_classifier_config" if legacy else "opensource_classifier_config": {
"provider": "databricks",
"model": "ai_decide",
**({"api_key": "dapi-configured", "api_base": api_base} if configured_key else {}),
},
"tiers": {"SIMPLE": "cheap"},
},
derive_savings_baseline=False,
)
with respx.mock(assert_all_called=True) as upstream:
route: Final = upstream.post("https://workspace.test/api/2.0/ai-functions/ai-decide").respond(
200,
json={"response": {"answers": {"tier": _answer().model_dump()}}, "metadata": {"version": "1.0"}},
)
outcome: Final = await router.aclassify("choose a tier")
await GLOBAL_LOGGING_WORKER.flush()
assert outcome.cause == "jev_classifier"
assert outcome.jev_verdict is not None
assert (outcome.jev_verdict.provider, outcome.jev_verdict.model) == ("databricks", "ai_decide")
assert outcome.jev_verdict.label == "SIMPLE"
assert outcome.classifier_cost is None
sent: Final = route.calls.last.request
assert sent.headers.get("authorization") == ("Bearer dapi-configured" if configured_key else "Bearer dapi-env-token")
assert set(json.loads(sent.content)) == {"state", "questions"}
assert len(recorder.calls) == 1
def test_databricks_without_a_base_or_key_fails_at_client_build(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("DATABRICKS_API_BASE", raising=False)
monkeypatch.delenv("DATABRICKS_API_KEY", raising=False)
@ -618,7 +670,7 @@ def test_databricks_without_a_base_or_key_fails_at_client_build(monkeypatch: pyt
litellm.Router(model_list=[]),
{
"classifier_type": "oss_classifier",
"opensource_classifier_config": {"provider": "databricks", "model": "databricks-openjev-qwen35-4b"},
"opensource_classifier_config": {"provider": "databricks", "model": "ai_decide"},
"tiers": {"SIMPLE": "cheap"},
},
derive_savings_baseline=False,
@ -665,17 +717,12 @@ def test_jev_api_base_without_its_own_key_is_rejected_so_the_environment_key_sta
def test_databricks_api_base_without_its_own_key_is_rejected_so_the_workspace_token_stays_home() -> None:
with pytest.raises(ValueError, match="DATABRICKS_API_KEY or DATABRICKS_TOKEN is only sent to DATABRICKS_API_BASE"):
JevClassifierConfig.model_validate(
{"provider": "databricks", "model": "my-openjev", "api_base": "https://collector.invalid/serving-endpoints"}
{"provider": "databricks", "model": "ai_decide", "api_base": "https://collector.invalid"}
)
paired: Final = JevClassifierConfig.model_validate(
{
"provider": "databricks",
"model": "my-openjev",
"api_base": "https://workspace.invalid/serving-endpoints",
"api_key": "dapi-own",
}
{"provider": "databricks", "model": "ai_decide", "api_base": "https://workspace.invalid", "api_key": "dapi-own"}
)
assert (paired.api_base, paired.api_key) == ("https://workspace.invalid/serving-endpoints", "dapi-own")
assert (paired.api_base, paired.api_key) == ("https://workspace.invalid", "dapi-own")
@pytest.mark.parametrize(

View file

@ -55,9 +55,7 @@ const ClassifierTypeRadios: React.FC<ClassifierTypeRadiosProps> = ({ value, clas
<RadioGroupItem value="jev" className="mt-0.5" />
<span>
<strong className="font-semibold">OSS Classifier</strong>{" "}
<span className="text-muted-foreground">
uses Jev, Laya, Bespoke, or a Databricks endpoint to decide the tier
</span>
<span className="text-muted-foreground">uses Jev, Laya, Bespoke, or Databricks to decide the tier</span>
</span>
</Label>
<SimpleTooltip content={scorerLockedReason}>

View file

@ -125,7 +125,7 @@ describe("JEV classifier editor", () => {
expect(screen.getByLabelText("Classifier Model")).toHaveValue("jev-latest");
await chooseSelectOption(userEvent, screen.getByRole("combobox", { name: "Decision model" }), label);
if (provider === "bespoke") expect(screen.getByLabelText("Classifier Model")).toHaveTextContent("nimble-latest");
if (provider === "databricks") expect(screen.getByLabelText("Classifier Model")).toHaveValue("");
if (provider === "databricks") expect(screen.getByLabelText("Classifier Model")).toHaveValue("ai_decide");
if (typesTheModel) {
fireEvent.change(screen.getByLabelText("Classifier Model"), { target: { value: model } });
} else {

View file

@ -15,7 +15,7 @@ const providerDescriptions = {
bespoke:
"Uses Bespoke Nimble with your configured tiers. Set BESPOKE_API_BASE on the gateway to connect your Nimble server.",
databricks:
"Uses a Databricks ai_decide serving endpoint with your configured tiers. Set DATABRICKS_API_BASE and DATABRICKS_API_KEY on the gateway, and enter the serving endpoint name as the classifier model.",
"Uses the Databricks ai_decide AI Function, or a decision serving endpoint you name, with your configured tiers. Set DATABRICKS_API_BASE to https://<workspace-host>/serving-endpoints and DATABRICKS_API_KEY on the gateway.",
};
export default function JevClassifierConfig({
@ -59,7 +59,7 @@ export default function JevClassifierConfig({
value={config.model}
placeholder={
config.provider === "databricks"
? "Serving endpoint name, e.g. databricks-openjev-qwen35-4b"
? "ai_decide, or a serving endpoint name such as databricks-openjev-qwen35-4b"
: undefined
}
onChange={(event) => update({ model: event.target.value })}

View file

@ -54,6 +54,8 @@ describe("buildAutoRouterRoutingTestRequest", () => {
["bespoke", "bespokelabs/Bespoke-Nimble-9B", "json"],
["databricks", "databricks-openjev-qwen35-4b", "object"],
["databricks", "databricks-openjev-qwen35-4b", "json"],
["databricks", "ai_decide", "object"],
["databricks", "ai_decide", "json"],
])("probes saved %s/%s %s configuration with custom tiers and team context", (provider, model, format) => {
const config = {
classifier_type: "oss_classifier",

View file

@ -163,6 +163,7 @@ describe("buildComplexityRouterConfig", () => {
["bespoke", "nimble"],
["bespoke", "bespokelabs/Bespoke-Nimble-9B"],
["databricks", "databricks-openjev-qwen35-4b"],
["databricks", "ai_decide"],
["jev", "custom-jev-model"],
[undefined, "custom-jev-model"],
] as const)("accepts %s model %s before saving or testing", (provider, model) => {

View file

@ -27,11 +27,11 @@ describe("hydrateOssClassifier", () => {
});
describe("jevClassifierConfigSchema", () => {
it("starts a Databricks classifier with an empty endpoint name that does not pass until typed", () => {
it("starts a Databricks classifier on ai_decide and still takes a typed serving endpoint name", () => {
const fresh = defaultJevClassifierConfig("databricks");
expect(fresh).toEqual({ provider: "databricks", model: "", timeout_ms: 3000 });
expect(jevClassifierConfigSchema.safeParse(fresh).success).toBe(false);
expect(fresh).toEqual({ provider: "databricks", model: "ai_decide", timeout_ms: 3000 });
expect(jevClassifierConfigSchema.safeParse(fresh).success).toBe(true);
expect(jevClassifierConfigSchema.safeParse({ ...fresh, model: "my-openjev" }).success).toBe(true);
});

View file

@ -13,12 +13,13 @@ export const isOssClassifierProvider = (value: unknown): value is OssClassifierP
const DEFAULT_TIMEOUT_MS = 3000;
const SERVING_ENDPOINT_NAME = /^[A-Za-z0-9_-][A-Za-z0-9._-]*$/;
const DATABRICKS_AI_DECIDE_MODEL = "ai_decide";
export const fixedClassifierModels = (provider: OssClassifierProvider | undefined): readonly string[] | undefined =>
provider === "laya" || provider === "bespoke" ? OSS_CLASSIFIER_MODELS[provider] : undefined;
const defaultClassifierModel = (provider: OssClassifierProvider | undefined): string =>
fixedClassifierModels(provider)?.[0] ?? (provider === "databricks" ? "" : "jev-latest");
fixedClassifierModels(provider)?.[0] ?? (provider === "databricks" ? DATABRICKS_AI_DECIDE_MODEL : "jev-latest");
const isSupportedClassifierModel = (provider: OssClassifierProvider | undefined, model: string): boolean => {
if (provider === "databricks") return SERVING_ENDPOINT_NAME.test(model);
@ -49,7 +50,7 @@ export const jevClassifierConfigSchema = z
.refine(
(config) => config.deployment_name !== undefined || isSupportedClassifierModel(config.provider, config.model),
{
error: "Select a supported classifier model, or enter the bare Databricks serving endpoint name",
error: "Select a supported classifier model, or enter ai_decide or a bare Databricks serving endpoint name",
path: ["model"],
},
);
@ -57,9 +58,7 @@ export const jevClassifierConfigSchema = z
export type JevClassifierConfig = z.infer<typeof jevClassifierConfigSchema>;
export const defaultJevClassifierConfig = (provider: OssClassifierProvider = "jev"): JevClassifierConfig =>
provider === "databricks"
? { provider, model: "", timeout_ms: DEFAULT_TIMEOUT_MS }
: jevClassifierConfigSchema.parse({ provider });
jevClassifierConfigSchema.parse({ provider });
export const hydrateOssClassifier = (config: {
classifier_type?: ClassifierType | "oss_classifier";