Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/external_router/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,7 @@ pub mod metrics_labels {
pub const ENDPOINT_CHAT: &str = "chat";
pub const ENDPOINT_GENERATE: &str = "generate";
pub const ENDPOINT_RESPONSES: &str = "responses";
pub const ENDPOINT_DECISIONS: &str = "decisions";
pub const ENDPOINT_COMPLETIONS: &str = "completions";
pub const ENDPOINT_RERANK: &str = "rerank";
pub const ENDPOINT_EMBEDDINGS: &str = "embeddings";
Expand Down
155 changes: 155 additions & 0 deletions e2e_test/router/test_decisions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,155 @@
"""OpenAI Decisions through a regular HTTP gateway and real SGLang inference.

Uses raw HTTP because the installed OpenAI SDK can predate Decisions. The
SGLang adapter performs prefill scoring, so every response has zero output
usage. Run with E2E_ENGINE=sglang E2E_RUNTIME=sglang pytest
e2e_test/router/test_decisions.py -v.
"""

from __future__ import annotations

import math

import httpx
import pytest

_MODEL = "Qwen/Qwen3-4B-Instruct-2507"


def _post(gateway, body: dict) -> httpx.Response:
return httpx.post(f"{gateway.base_url}/v1/decisions", json=body, timeout=120.0)


def _assert_distribution(answer: dict) -> None:
probabilities = [item["probability"] for item in answer["probabilities"]]
assert probabilities
assert all(math.isfinite(p) and 0 <= p <= 1 for p in probabilities)
assert sum(probabilities) == pytest.approx(1.0, abs=1e-6)
assert math.isfinite(answer["confidence"])
assert 0 <= answer["confidence"] <= 1


@pytest.mark.engine("sglang")
@pytest.mark.gpu(1)
@pytest.mark.e2e
@pytest.mark.model(_MODEL)
@pytest.mark.parametrize("setup_backend", ["http"], indirect=True)
class TestDecisions:
def test_all_question_types_order_names_and_prefill_usage(self, setup_backend):
_, model, _, gateway = setup_backend
response = _post(
gateway,
{
"model": model,
"input": "The sky is blue. Two plus two is four. The review is excellent.",
"questions": [
{"type": "predicate", "instructions": "Is two plus two four?"},
{
"type": "choice",
"name": "repeated",
"instructions": "What color is the sky?",
"choices": [{"value": "blue"}, {"value": "red"}],
},
{
"type": "score",
"name": "repeated",
"instructions": "Rate the review quality.",
"levels": [
{"label": "poor", "description": "Bad quality"},
{"label": "excellent", "description": "Very good quality"},
],
},
{
"type": "choice",
"instructions": "Is the sky blue?",
"choices": [
{"value": True, "description": "Yes, the sky is blue"},
{"value": False, "description": "No, it is another color"},
],
},
],
},
)
assert response.status_code == 200, response.text
assert response.headers["content-type"].startswith("application/json")
result = response.json()
answers = result["answers"]
assert [a["type"] for a in answers] == ["predicate", "choice", "score", "choice"]
assert [a["name"] for a in answers] == [None, "repeated", "repeated", None]
assert 0.5 < answers[0]["probability"] <= 1
assert answers[1]["choice"] == "blue"
assert [p["value"] for p in answers[1]["probabilities"]] == ["blue", "red"]
assert [p["value"] for p in answers[2]["probabilities"]] == [0, 1]
assert [p["label"] for p in answers[2]["probabilities"]] == ["poor", "excellent"]
# Scores are expectations over level indices, not the winning level.
assert answers[2]["score"] == pytest.approx(
sum(p["value"] * p["probability"] for p in answers[2]["probabilities"]),
abs=1e-6,
)
assert type(answers[3]["choice"]) is bool
assert [p["value"] for p in answers[3]["probabilities"]] == [True, False]
assert all(type(p["value"]) is bool for p in answers[3]["probabilities"])
for answer in answers[1:]:
_assert_distribution(answer)
usage = result["usage"]
assert usage["input_tokens"] > 0
assert usage["output_tokens"] == 0
assert usage["total_tokens"] == usage["input_tokens"]
assert usage["output_tokens_details"]["reasoning_tokens"] == 0
assert set(usage["input_tokens_details"]) >= {"cached_tokens", "cache_write_tokens"}

def test_ordered_user_messages_are_accepted(self, setup_backend):
_, model, _, gateway = setup_backend
response = _post(
gateway,
{
"model": model,
"input": [
{"role": "user", "content": "The delivery arrived."},
{
"role": "user",
"content": [{"type": "input_text", "text": "The package is intact."}],
},
],
"questions": [
{"type": "predicate", "instructions": "Did the delivery arrive intact?"}
],
},
)
assert response.status_code == 200, response.text
answer = response.json()["answers"][0]
assert answer["type"] == "predicate"
assert answer["name"] is None
assert 0.5 < answer["probability"] <= 1

@pytest.mark.parametrize(
"invalid", ["missing_model", "unknown_type", "numeric_choice", "image"]
)
def test_invalid_or_unsupported_requests_are_client_errors(self, setup_backend, invalid):
_, model, _, gateway = setup_backend
body = {
"model": model,
"input": "A short text.",
"questions": [{"type": "predicate", "instructions": "Is this text?"}],
}
if invalid == "missing_model":
del body["model"]
elif invalid == "unknown_type":
body["questions"][0]["type"] = "unknown"
elif invalid == "numeric_choice":
body["questions"] = [
{"type": "choice", "instructions": "Choose", "choices": [{"value": 1}]}
]
else:
# Valid OpenAI image input; SGLang's pinned text-only adapter rejects it.
body["input"] = [
{
"role": "user",
"content": [{"type": "input_image", "image_url": "data:image/png;base64,AA=="}],
}
]
response = _post(gateway, body)
assert 400 <= response.status_code < 500, response.text
if invalid == "image":
assert response.status_code == 400, response.text
assert isinstance(response.json()["error"]["message"], str)
133 changes: 133 additions & 0 deletions model_gateway/src/routers/gateway.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ use openai_protocol::{
chat::ChatCompletionRequest,
classify::ClassifyRequest,
completion::CompletionRequest,
decisions::DecisionsRequest,
embedding::EmbeddingRequest,
generate::GenerateRequest,
interactions::InteractionsRequest,
Expand Down Expand Up @@ -588,6 +589,43 @@ impl RouterTrait for Gateway {
.await
}

async fn route_decisions(
&self,
headers: Option<&HeaderMap>,
tenant_meta: &TenantRequestMeta,
body: DecisionsRequest,
model_id: &str,
) -> Response {
let router = if self.enable_igw {
let snapshot = self.worker_registry.get_routing_snapshot(model_id);
if snapshot.is_empty() {
return route_error::model_not_found(model_id);
}
// Decisions is supported only by regular HTTP workers. General
// weighted dispatch could send the same model to PD or gRPC.
if snapshot.pool(RoutingPool::HttpRegular).is_empty() {
return route_error::not_implemented(
"decisions_not_supported",
"Decisions requires a regular HTTP worker for this model",
);
}
self.routers.load().get(&router_ids::HTTP_REGULAR).cloned()
} else {
self.select_router_for_request(Some(model_id))
};
match router {
Some(router) => {
router
.route_decisions(headers, tenant_meta, body, model_id)
.await
}
None => route_error::not_implemented(
"decisions_not_supported",
"Decisions requires a regular HTTP router",
),
}
}

async fn route_audio_transcriptions(
&self,
headers: Option<&HeaderMap>,
Expand Down Expand Up @@ -879,6 +917,101 @@ mod tests {
RouteRequestMeta::new(TenantKey::from("test-tenant"))
}

#[derive(Debug)]
struct DecisionsStubRouter;

#[async_trait]
impl RouterTrait for DecisionsStubRouter {
fn as_any(&self) -> &dyn std::any::Any {
self
}

async fn route_decisions(
&self,
_headers: Option<&HeaderMap>,
_tenant_meta: &TenantRequestMeta,
_body: DecisionsRequest,
_model_id: &str,
) -> Response {
StatusCode::OK.into_response()
}

fn router_type(&self) -> &'static str {
"decisions"
}
}

async fn decision_status(gateway: &Gateway, model: &str) -> StatusCode {
let body = serde_json::from_value(serde_json::json!({
"model": model, "input": "evidence",
"questions": [{"type": "predicate", "instructions": "Is it evidence?"}]
}))
.unwrap();
gateway
.route_decisions(None, &test_tenant_meta(), body, model)
.await
.status()
}

#[tokio::test]
async fn decisions_use_only_regular_http_workers_in_mixed_fleets() {
for has_http_regular in [true, false] {
let gateway = test_gateway(true);
gateway.register_router(router_ids::HTTP_REGULAR, Arc::new(DecisionsStubRouter));
gateway.register_router(router_ids::HTTP_PD, Arc::new(PdStubRouter));
gateway.register_router(router_ids::GRPC_REGULAR, Arc::new(PdStubRouter));
gateway.set_default_router(router_ids::GRPC_REGULAR);
let workers = [
(ConnectionMode::Http, WorkerType::Regular),
(ConnectionMode::Http, WorkerType::Prefill),
(ConnectionMode::Http, WorkerType::Decode),
(ConnectionMode::Grpc, WorkerType::Regular),
(ConnectionMode::Zmq, WorkerType::Regular),
];
for (index, (mode, role)) in workers.into_iter().enumerate() {
if index == 0 && !has_http_regular {
continue;
}
gateway
.worker_registry
.register(Arc::new(
BasicWorkerBuilder::new(format!("http://worker-{index}:8080"))
.connection_mode(mode)
.worker_type(role)
.model(ModelCard::new("m").with_alias("alias"))
.build(),
))
.unwrap();
}
let expected = if has_http_regular {
StatusCode::OK
} else {
StatusCode::NOT_IMPLEMENTED
};
for _ in 0..32 {
assert_eq!(decision_status(&gateway, "m").await, expected);
assert_eq!(decision_status(&gateway, "alias").await, expected);
}
assert_eq!(
decision_status(&gateway, "missing").await,
StatusCode::NOT_FOUND
);
}
}

#[tokio::test]
async fn decisions_respect_single_router_configuration() {
let gateway = test_gateway(false);
gateway.register_router(router_ids::HTTP_REGULAR, Arc::new(DecisionsStubRouter));
assert_eq!(decision_status(&gateway, "m").await, StatusCode::OK);
gateway.register_router(router_ids::HTTP_PD, Arc::new(PdStubRouter));
gateway.set_default_router(router_ids::HTTP_PD);
assert_eq!(
decision_status(&gateway, "m").await,
StatusCode::NOT_IMPLEMENTED
);
}

#[tokio::test]
async fn count_tokens_selects_http_workers_without_a_decode_leg() {
for (role, router_id) in [
Expand Down
1 change: 1 addition & 0 deletions model_gateway/src/routers/grpc/utils/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ pub(crate) fn route_to_endpoint(route: &str) -> &'static str {
"/v1/completions" => metrics_labels::ENDPOINT_COMPLETIONS,
"/v1/rerank" => metrics_labels::ENDPOINT_RERANK,
"/v1/responses" => metrics_labels::ENDPOINT_RESPONSES,
"/v1/decisions" => metrics_labels::ENDPOINT_DECISIONS,
"/v1/messages" => metrics_labels::ENDPOINT_MESSAGES,
"/v1/audio/transcriptions" => metrics_labels::ENDPOINT_AUDIO_TRANSCRIPTIONS,
_ => "other",
Expand Down
Loading
Loading