[Feat] Structured decisions endpoint (/v1/systemone) (#59299)

Signed-off-by: Matt Mastracci <matthew@mastracci.com>
Signed-off-by: Lucas Wilkinson <lwilkins@redhat.com>
Signed-off-by: Francisco Javier Arceo <4163062+franciscojavierarceo@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Lucas Wilkinson <lwilkins@redhat.com>
Co-authored-by: Lucas Wilkinson <LucasWilkinson@users.noreply.github.com>
Co-authored-by: Francisco Javier Arceo <4163062+franciscojavierarceo@users.noreply.github.com>
Co-authored-by: Chauncey <chaunceyjiang@gmail.com>
This commit is contained in:
Matt Mastracci
2026-10-08 14:48:12 +00:00
committed by GitHub
co-authored by Claude Opus 5.5 Lucas Wilkinson Lucas Wilkinson Francisco Javier Arceo Chauncey
parent 273e099693
commit d33872dfeb
17 changed files with 1285 additions and 61 deletions
+3
View File
@@ -76,6 +76,9 @@ For further details on speech to text, please refer to [this page](speech_to_tex
- [Generative Scoring API](generative_scoring.md) (`/generative_scoring`)
- Applicable to [CausalLM models](../../models/generative_models.md) (task `"generate"`).
- Computes next-token probabilities for specified `label_token_ids`.
- [Structured Decisions API](structured_decisions.md) (`/v1/systemone`)
- Applicable to [text generation models](../../models/generative_models.md) with a [chat template](#chat-template).
- Answers typed questions about a state with a probability for every allowed answer.
## Instrumentator APIs
@@ -0,0 +1,110 @@
# Structured Decisions
The `/v1/systemone` endpoint answers a set of typed questions about a state and
returns a probability for every allowed answer.
The endpoint is off unless the server starts with
`--enable-structured-decisions`. With the flag, startup fails for a model that
does not support structured decisions, and with `--logprobs-mode raw_logits` or
`processed_logits`, since `label_mass` needs log probabilities.
## How it works
1. Each question is one read. The user message is the state, followed by the
question and its options, labeled `A`, `B`, `C`, ... in order, and
"Answer with the letter of one option only."
2. The read is the reply's first token. Each label must be one token there,
or the request returns 400.
3. The read returns the logprobs of that question's label tokens. Softmax over
the labels gives the answer's probabilities.
Thinking is off unless `chat_template_kwargs` turns it on.
This endpoint does not serve diffusion models yet.
Every read of a request starts with the state, so with
`--enable-prefix-caching` the state is prefilled once and each further
question prefills only its own text, then reads one token.
## Question types
| type | criteria | answer |
| --- | --- | --- |
| `choice` | map of option name to a description or `null` | `choice`, `probabilities` by option name, `confidence` |
A choice has at least one option. A question id is a non-empty string.
## Example
```bash
curl -s http://localhost:8000/v1/systemone \
-H "Content-Type: application/json" \
-d '{
"model": "Qwen/Qwen3-0.6B",
"state": {"ticket": "My card was charged twice for one order."},
"questions": {
"team": {
"type": "choice",
"instructions": "Which team should handle this ticket?",
"criteria": {
"billing": "payments and refunds",
"shipping": "deliveries",
"security": "account access"
}
}
},
"chat_template_kwargs": {"enable_thinking": false}
}'
```
Response, with probabilities that depend on the model:
```json
{
"id": "decision-...",
"object": "structured_decision",
"created": 1790000000,
"model": "Qwen/Qwen3-0.6B",
"answers": {
"team": {
"type": "choice",
"choice": "billing",
"probabilities": {"billing": 0.97, "shipping": 0.01, "security": 0.02},
"confidence": 0.90
}
},
"usage": {"input_tokens": 142, "output_tokens": 1},
"diagnostics": {
"team": {"label_mass": 0.93, "argmax_is_label": true}
}
}
```
`diagnostics.label_mass` is the probability the model put on the labels across
its whole vocabulary. A low value means most of the probability went to tokens
that are not labels. `confidence` is the chosen option's probability over the
whole vocabulary: its share of the labels times `label_mass`.
## Labels
Options are labeled `A` to `Z` in the order the request lists them, so a
`choice` has at most 26 options.
## Limits
| limit | value |
| --- | --- |
| questions per request | 64 |
| options per `choice` | 26 |
## Request fields
| field | meaning |
| --- | --- |
| `state` | what the questions are about: a string, or JSON that is sent as its JSON text |
| `questions` | question id to `{type, instructions, criteria}`, asked in this order |
| `instructions` | optional context placed ahead of the questions |
| `chat_template_kwargs` | passed to the chat template, for example `{"enable_thinking": false}` |
A question with any field other than `type`, `instructions` and `criteria` is
rejected with a 400.
+1
View File
@@ -175,6 +175,7 @@ When `--api-key` is configured, the following endpoints require Bearer token aut
- `/v1/responses/{response_id}/cancel` - Cancel a response
- `/v1/score` - Scoring API
- `/v1/rerank` - Reranking API
- `/v1/systemone` - Structured decisions API
- `/v1/load_lora_adapter` - Load a LoRA adapter (can alter model behavior; only available when `--enable-lora` is set and `VLLM_ALLOW_RUNTIME_LORA_UPDATING=True`)
- `/v1/unload_lora_adapter` - Unload a LoRA adapter (can alter model behavior; only available when `--enable-lora` is set and `VLLM_ALLOW_RUNTIME_LORA_UPDATING=True`)
- `/inference/v1/generate` - Generate completions (available when `--enable-scale-out` is set, or with `--tokens-only`)
@@ -0,0 +1,105 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from typing import Any
import pytest
from vllm.entrypoints.generate.structured_decisions.question_types import (
QUESTION_TYPES,
Option,
Question,
QuestionType,
StructuredDecisionError,
build_question,
register_question_type,
)
def choice(qid="bucket", criteria=None, instructions="Which team?", max_options=128):
return build_question(
qid,
"choice",
instructions,
criteria or {"billing": "money", "outage": None, "other": None},
max_options,
)
def test_choice_labels_follow_option_order():
q = choice()
assert q.labels == ("A", "B", "C")
assert [a.name for a in q.options] == ["billing", "outage", "other"]
assert q.options[0].description == "money"
wide = choice(criteria={str(i): None for i in range(26)})
assert wide.labels[-1] == "Z"
def test_choice_prompt():
q = choice()
assert q.type.prompt(q) == (
"Question: Which team?\n"
"A: billing - money\n"
"B: outage\n"
"C: other\n"
"Answer with the letter of one option only."
)
def test_choice_answer_shape():
q = choice()
answer = q.type.answer(q, [0.2, 0.7, 0.1], 0.5)
assert answer == {
"type": "choice",
"choice": "outage",
"probabilities": {"billing": 0.2, "outage": 0.7, "other": 0.1},
"confidence": 0.35,
}
@pytest.mark.parametrize(
"qid,type_name,criteria,match",
[
("", "choice", {"x": None, "y": None}, "non-empty"),
("q", "nope", {"x": None, "y": None}, "unknown question type"),
("q", "choice", ["x", "y"], "must map option names"),
],
)
def test_build_question_rejects(qid, type_name, criteria, match):
with pytest.raises(StructuredDecisionError, match=match):
build_question(qid, type_name, "", criteria, 128)
def test_one_option_choice():
q = choice(criteria={"only": None})
assert q.labels == ("A",)
assert q.type.answer(q, [1.0], 0.4)["confidence"] == 0.4
def test_option_limit():
with pytest.raises(StructuredDecisionError, match="at most 2 options"):
choice(criteria={"x": None, "y": None, "z": None}, max_options=2)
with pytest.raises(StructuredDecisionError, match="at most 26 options"):
choice(criteria={str(i): None for i in range(27)})
def test_registered_type_plugs_in():
class BinaryQuestion(QuestionType):
name = "test_binary"
def parse_options(self, qid: str, criteria: Any) -> list[Option]:
return [Option("yes"), Option("no")]
def answer(
self, question: Question, probs: list[float], label_mass: float
) -> dict[str, Any]:
return {"type": self.name, "yes": probs[0]}
register_question_type(BinaryQuestion)
try:
q = build_question("ok", "test_binary", "Is it fine?", None, 128)
assert q.labels == ("A", "B")
assert q.type.answer(q, [0.9, 0.1], 1.0) == {"type": "test_binary", "yes": 0.9}
with pytest.raises(ValueError, match="already registered"):
register_question_type(BinaryQuestion)
finally:
del QUESTION_TYPES["test_binary"]
@@ -0,0 +1,65 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from types import SimpleNamespace
from typing import Any
import pytest
from fastapi import FastAPI
from vllm.entrypoints.generate.structured_decisions.api_router import (
register_structured_decisions_api_router,
)
from vllm.entrypoints.generate.structured_decisions.question_types import LABELS
from vllm.entrypoints.generate.structured_decisions.strategies import (
NextTokenStrategy,
reply_label_ids,
select_read_strategy,
)
def model(architecture: str, logprobs_mode: str = "raw_logprobs") -> Any:
return SimpleNamespace(architecture=architecture, logprobs_mode=logprobs_mode)
def test_strategy_selection():
qwen = "Qwen3ForCausalLM"
assert select_read_strategy(model(qwen)) is NextTokenStrategy
assert select_read_strategy(model(qwen, "processed_logprobs")) is NextTokenStrategy
with pytest.raises(ValueError, match="does not support LlamaForCausalLM"):
select_read_strategy(model("LlamaForCausalLM"))
with pytest.raises(ValueError, match="not raw_logits"):
select_read_strategy(model(qwen, "raw_logits"))
def test_route_needs_the_flag():
for enabled in (False, True):
app = FastAPI()
app.state.args = SimpleNamespace(enable_structured_decisions=enabled)
register_structured_decisions_api_router(app)
paths = {getattr(route, "path", None) for route in app.routes}
assert ("/v1/systemone" in paths) == enabled
@pytest.fixture(scope="module")
def qwen():
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-0.6B")
prompt_ids = tokenizer.apply_chat_template(
[{"role": "user", "content": "state"}],
add_generation_prompt=True,
enable_thinking=False,
tokenize=True,
return_dict=False,
)
return tokenizer, prompt_ids
def test_labels_start_the_reply(qwen):
tokenizer, prompt_ids = qwen
tail, ids = reply_label_ids(tokenizer, prompt_ids)
assert tokenizer.decode(tail) == "\n\n"
assert [tokenizer.decode([i]) for i in ids] == list(LABELS)
# After a colon, Qwen writes ":A" as one token, so "A" is not one token.
with pytest.raises(ValueError, match="not one distinct token"):
reply_label_ids(tokenizer, tokenizer.encode("team:"))
@@ -0,0 +1,217 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import pytest
import requests
from tests.utils import RemoteOpenAIServer
MODEL_NAME = "Qwen/Qwen3-0.6B"
@pytest.fixture(scope="module")
def server():
args = [
"--dtype",
"bfloat16",
"--max-model-len",
"1024",
"--enforce-eager",
"--max-num-seqs",
"32",
"--enable-prefix-caching",
"--enable-structured-decisions",
]
with RemoteOpenAIServer(MODEL_NAME, args) as remote_server:
yield remote_server
def post(server, body):
return requests.post(server.url_for("v1/systemone"), json=body)
def choice(instructions, *names):
return {
"type": "choice",
"instructions": instructions,
"criteria": dict.fromkeys(names),
}
COPY = {
"team": choice(
"Which team does the message name?", "billing", "shipping", "security"
),
"lang": choice("Which language does the message name?", "English", "French"),
}
def test_choice_decision(server):
response = post(
server,
{
"model": MODEL_NAME,
"state": {"team": "billing", "language": "French"},
"questions": COPY,
"chat_template_kwargs": {"enable_thinking": False},
},
)
assert response.status_code == 200, response.text
body = response.json()
assert body["object"] == "structured_decision"
assert list(body["answers"]) == ["team", "lang"]
team = body["answers"]["team"]
assert team["type"] == "choice"
assert set(team["probabilities"]) == {"billing", "shipping", "security"}
assert sum(team["probabilities"].values()) == pytest.approx(1.0, abs=1e-4)
assert team["confidence"] == pytest.approx(
max(team["probabilities"].values()) * body["diagnostics"]["team"]["label_mass"]
)
for diag in body["diagnostics"].values():
assert 0.0 < diag["label_mass"] <= 1.0 + 1e-6
assert body["usage"]["output_tokens"] == 2
assert body["usage"]["input_tokens"] > 0
def test_one_option_choice(server):
response = post(
server,
{
"model": MODEL_NAME,
"state": "Team: billing.",
"questions": {
"team": choice("Which team does the message name?", "billing")
},
"chat_template_kwargs": {"enable_thinking": False},
},
)
assert response.status_code == 200, response.text
body = response.json()
team = body["answers"]["team"]
assert team["choice"] == "billing"
assert team["probabilities"] == {"billing": 1.0}
assert team["confidence"] == pytest.approx(
body["diagnostics"]["team"]["label_mass"]
)
def test_one_option_coin_flip(server):
# A fair coin with only "heads" allowed. The answer is forced, so its
# confidence must come from the model's probability of the label.
response = post(
server,
{
"model": MODEL_NAME,
"state": "I flip a fair coin.",
"questions": {"coin": choice("How did the coin land?", "heads")},
"chat_template_kwargs": {"enable_thinking": False},
},
)
assert response.status_code == 200, response.text
body = response.json()
coin = body["answers"]["coin"]
assert coin["probabilities"] == {"heads": 1.0}
assert coin["confidence"] == pytest.approx(
body["diagnostics"]["coin"]["label_mass"]
)
@pytest.mark.parametrize(
"state,questions,expected",
[
(
{"team": "billing", "language": "French"},
COPY,
{"team": "billing", "lang": "French"},
),
(
"My card was charged twice for one order.",
{
"card": choice("Does the message mention a card?", "yes", "no"),
"mood": choice("How does the customer feel?", "happy", "angry"),
},
{"card": "yes", "mood": "angry"},
),
(
"Hello.",
{
"sky": choice(
"What color is the sky on a clear day?", "blue", "green", "red"
)
},
{"sky": "blue"},
),
],
)
def test_single_read_answers(server, state, questions, expected):
response = post(
server,
{
"model": MODEL_NAME,
"state": state,
"questions": questions,
"chat_template_kwargs": {"enable_thinking": False},
},
)
assert response.status_code == 200, response.text
answers = response.json()["answers"]
assert {qid: a["choice"] for qid, a in answers.items()} == expected
def test_option_limit(server):
body = {
"model": MODEL_NAME,
"state": "Pick option 7.",
"chat_template_kwargs": {"enable_thinking": False},
}
ok = post(
server, {**body, "questions": {"n": choice("Which?", *map(str, range(26)))}}
)
assert ok.status_code == 200, ok.text
assert len(ok.json()["answers"]["n"]["probabilities"]) == 26
over = post(
server, {**body, "questions": {"n": choice("Which?", *map(str, range(27)))}}
)
assert over.status_code == 400
assert "at most 26 options" in over.json()["error"]["message"]
def test_thinking_is_refused(server):
response = post(
server,
{
"model": MODEL_NAME,
"state": "x",
"questions": {"q": choice("Which?", "a", "b")},
"chat_template_kwargs": {"enable_thinking": True},
},
)
assert response.status_code == 400
assert "thinking must be off" in response.json()["error"]["message"]
@pytest.mark.parametrize(
"questions,match",
[
(
{
"q": {
"type": "choice",
"criteria": {"a": None, "b": None},
"depends_on": ["x"],
}
},
"unknown field(s)",
),
(
{
f"q{i}": {"type": "choice", "criteria": {"a": None, "b": None}}
for i in range(65)
},
"at most 64",
),
],
)
def test_validation_errors(server, questions, match):
response = post(server, {"model": MODEL_NAME, "state": "x", "questions": questions})
assert response.status_code == 400
assert match in response.json()["error"]["message"]
+33
View File
@@ -53,6 +53,12 @@ def register_generate_api_routers(app: FastAPI):
register_generative_scoring_api_router(app)
from .structured_decisions.api_router import (
register_structured_decisions_api_router,
)
register_structured_decisions_api_router(app)
async def init_generate_state(
engine_client: "EngineClient",
@@ -220,3 +226,30 @@ async def init_generate_state(
state.openai_serving_models,
request_logger=request_logger,
)
from .structured_decisions.serving import ServingStructuredDecisions
from .structured_decisions.strategies import ReadContext, select_read_strategy
strategy_cls = (
select_read_strategy(engine_client.model_config)
if "generate" in supported_tasks
and getattr(args, "enable_structured_decisions", False)
else None
)
state.serving_structured_decisions = (
ServingStructuredDecisions(
state.openai_serving_models,
strategy_cls(
ReadContext(
engine_client=engine_client,
online_renderer=state.online_renderer,
chat_template=resolved_chat_template,
chat_template_content_format=args.chat_template_content_format,
default_chat_template_kwargs=default_chat_template_kwargs,
)
),
request_logger=request_logger,
)
if strategy_cls is not None
else None
)
@@ -11,13 +11,14 @@ logits (task="generate").
import asyncio
import math
import time
from collections.abc import AsyncGenerator, Mapping
from collections.abc import Mapping
from typing import Literal
from fastapi import Request
from pydantic import Field
from vllm.engine.protocol import EngineClient
from vllm.entrypoints.generate.label_reads import next_token_label_reads
from vllm.entrypoints.openai.models.serving import OpenAIServingModels
from vllm.entrypoints.serve.engine.protocol import (
ErrorResponse,
@@ -28,7 +29,6 @@ from vllm.entrypoints.serve.engine.serving import BaseServing
from vllm.entrypoints.serve.utils.request_logger import RequestLogger
from vllm.inputs import EngineInput, tokens_input
from vllm.logger import init_logger
from vllm.outputs import RequestOutput
from vllm.sampling_params import SamplingParams
from vllm.tokenizers import TokenizerLike
from vllm.tracing import (
@@ -37,7 +37,6 @@ from vllm.tracing import (
log_tracing_disabled_warning,
)
from vllm.utils import random_uuid
from vllm.utils.async_utils import merge_async_iterators
logger = init_logger(__name__)
@@ -271,37 +270,28 @@ class ServingGenerativeScoring(BaseServing):
else await self._get_trace_headers(raw_request.headers)
)
# Schedule requests for all inputs
generators: list[AsyncGenerator[RequestOutput, None]] = []
for i, engine_input in enumerate(engine_inputs):
request_id_item = f"{request_id}-{i}"
self._log_inputs(
request_id_item,
f"{request_id}-{i}",
engine_input,
params=sampling_params,
lora_request=lora_request,
)
generator = self.engine_client.generate(
engine_input,
sampling_params,
request_id_item,
try:
reads = await next_token_label_reads(
self.engine_client,
engine_inputs,
[sampling_params] * len(engine_inputs),
request_id,
lora_request=lora_request,
trace_headers=trace_headers,
priority=request.priority,
)
generators.append(generator)
# Collect results
result_generator = merge_async_iterators(*generators)
results: list[RequestOutput | None] = [None] * len(engine_inputs)
try:
async for i, res in result_generator:
results[i] = res
except asyncio.CancelledError:
return self.create_error_response("Client disconnected")
except ValueError as e:
return self.create_error_response(e)
except Exception as e:
logger.exception("Error during generation")
return self.create_error_response(e)
@@ -311,45 +301,8 @@ class ServingGenerativeScoring(BaseServing):
total_prompt_tokens = 0
total_completion_tokens = 0
for i, result in enumerate(results):
if result is None:
return self.create_error_response(
f"Failed to generate result for item {i}"
)
# Check for errors
if result.outputs and result.outputs[0].finish_reason == "error":
return self.create_error_response(f"Generation error for item {i}")
# Get logprobs from the generated token
if not result.outputs or len(result.outputs) == 0:
return self.create_error_response(f"No output generated for item {i}")
output = result.outputs[0]
if output.logprobs is None or len(output.logprobs) == 0:
return self.create_error_response(
f"No logprobs available for item {i}. "
"This might indicate an issue with logprobs configuration."
)
# The logprobs dict maps token_id -> Logprob object
# For logprobs=-1, this contains all vocab tokens
logprobs_dict = output.logprobs[0]
# Extract logprobs for label tokens
label_logprobs: dict[int, float] = {}
missing_tokens = []
for token_id in request.label_token_ids:
if token_id in logprobs_dict:
label_logprobs[token_id] = logprobs_dict[token_id].logprob
else:
missing_tokens.append(token_id)
if missing_tokens:
return self.create_error_response(
f"Token IDs {missing_tokens} not found in logprobs for item {i}. "
"This might indicate the tokens are outside the model's vocabulary."
)
for i, read in enumerate(reads):
label_logprobs = dict(zip(request.label_token_ids, read.logprobs))
# Compute probabilities based on apply_softmax setting
token_probs = self._compute_probabilities(
@@ -370,7 +323,7 @@ class ServingGenerativeScoring(BaseServing):
# Update token counts
total_prompt_tokens += prompt_token_counts[i]
total_completion_tokens += len(output.token_ids)
total_completion_tokens += len(read.result.outputs[0].token_ids)
# Build response
model_name = self.models.model_name(lora_request)
+76
View File
@@ -0,0 +1,76 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""One-token reads that return the logprobs of chosen label tokens."""
from collections.abc import AsyncGenerator, Mapping, Sequence
from dataclasses import dataclass
from vllm.engine.protocol import EngineClient
from vllm.inputs import EngineInput
from vllm.lora.request import LoRARequest
from vllm.outputs import RequestOutput
from vllm.sampling_params import SamplingParams
from vllm.utils.async_utils import merge_async_iterators
@dataclass(frozen=True)
class LabelRead:
result: RequestOutput
logprobs: list[float]
"""Full-vocabulary logprob of each label token, in ``logprob_token_ids``
order."""
async def next_token_label_reads(
engine_client: EngineClient,
engine_inputs: Sequence[EngineInput],
sampling_params: Sequence[SamplingParams],
request_id: str,
*,
lora_request: LoRARequest | None = None,
trace_headers: Mapping[str, str] | None = None,
priority: int = 0,
) -> list[LabelRead]:
"""Runs one read per input at once, as request ``{request_id}-{i}``. Each
read's params set ``max_tokens=1`` and the label tokens as
``logprob_token_ids``. Raises ValueError naming the item when a read has
no output or lacks a label's logprob."""
generators: list[AsyncGenerator[RequestOutput, None]] = [
engine_client.generate(
engine_input,
params,
f"{request_id}-{i}",
lora_request=lora_request,
trace_headers=trace_headers,
priority=priority,
)
for i, (engine_input, params) in enumerate(zip(engine_inputs, sampling_params))
]
results: list[RequestOutput | None] = [None] * len(generators)
async for i, res in merge_async_iterators(*generators):
results[i] = res
reads = []
for i, (result, params) in enumerate(zip(results, sampling_params)):
if result is None:
raise ValueError(f"Failed to generate result for item {i}")
if not result.outputs:
raise ValueError(f"No output generated for item {i}")
output = result.outputs[0]
if output.finish_reason == "error":
raise ValueError(f"Generation error for item {i}")
if not output.logprobs:
raise ValueError(
f"No logprobs available for item {i}. "
"This might indicate an issue with logprobs configuration."
)
logprobs = output.logprobs[0]
label_ids = params.logprob_token_ids or []
missing = [t for t in label_ids if t not in logprobs]
if missing:
raise ValueError(
f"Token IDs {missing} not found in logprobs for item {i}. "
"This might indicate the tokens are outside the model's vocabulary."
)
reads.append(LabelRead(result, [logprobs[t].logprob for t in label_ids]))
return reads
@@ -0,0 +1,2 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
@@ -0,0 +1,53 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from http import HTTPStatus
from fastapi import APIRouter, Depends, FastAPI, Request
from fastapi.responses import JSONResponse
from vllm.entrypoints.serve.engine.protocol import ErrorResponse
from vllm.entrypoints.serve.utils.api_utils import (
load_aware_call,
validate_json_request,
with_cancellation,
)
from .protocol import StructuredDecisionRequest
from .serving import ServingStructuredDecisions
router = APIRouter()
def structured_decisions(request: Request) -> ServingStructuredDecisions | None:
return getattr(request.app.state, "serving_structured_decisions", None)
@router.post(
"/v1/systemone",
dependencies=[Depends(validate_json_request)],
responses={
HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
HTTPStatus.NOT_IMPLEMENTED.value: {"model": ErrorResponse},
},
)
@with_cancellation
@load_aware_call
async def create_structured_decision(
request: StructuredDecisionRequest, raw_request: Request
):
handler = structured_decisions(raw_request)
if handler is None:
raise NotImplementedError(
"The model does not support the structured decision API"
)
result = await handler.create_decision(request, raw_request)
if isinstance(result, ErrorResponse):
return JSONResponse(content=result.model_dump(), status_code=result.error.code)
return JSONResponse(content=result.model_dump())
def register_structured_decisions_api_router(app: FastAPI):
if getattr(app.state.args, "enable_structured_decisions", False):
app.include_router(router)
@@ -0,0 +1,88 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Request and response models of the structured decisions API (/v1/systemone)."""
import time
from typing import Any, Literal
from pydantic import Field
from vllm.config import ModelConfig
from vllm.entrypoints.chat_utils import ChatTemplateContentFormatOption
from vllm.entrypoints.serve.engine.protocol import OpenAIBaseModel
from vllm.renderers import ChatParams, TokenizeParams, merge_kwargs
from vllm.utils import random_uuid
class QuestionSpec(OpenAIBaseModel):
type: str
instructions: Any = ""
criteria: Any = None
class StructuredDecisionRequest(OpenAIBaseModel):
model: str | None = None
state: Any = Field(..., description="What the questions are about: text or JSON.")
questions: dict[str, QuestionSpec] = Field(
..., description="Question id to question, asked in this order."
)
instructions: str | None = Field(
default=None, description="Context placed ahead of the questions."
)
chat_template_kwargs: dict[str, Any] | None = None
priority: int = Field(default=0, ge=-(2**63), le=2**63 - 1)
request_id: str = Field(default_factory=random_uuid)
class DecisionUsage(OpenAIBaseModel):
input_tokens: int
output_tokens: int
class QuestionDiagnostics(OpenAIBaseModel):
label_mass: float = Field(
description="Probability the model put on the labels, over the full "
"vocabulary. A low value means most of the probability went to tokens "
"that are not labels."
)
argmax_is_label: bool
class StructuredDecisionResponse(OpenAIBaseModel):
id: str
object: Literal["structured_decision"] = "structured_decision"
created: int = Field(default_factory=lambda: int(time.time()))
model: str
answers: dict[str, dict[str, Any]]
usage: DecisionUsage
diagnostics: dict[str, QuestionDiagnostics]
class ReadPromptRequest(OpenAIBaseModel):
"""Chat options for one read's prompt: the state and the question in the
user turn, ending at the generation prompt so the label is the reply's
first token. Thinking is off unless the request turns it on, or the label
would follow a thought rather than start the reply."""
chat_template_kwargs: dict[str, Any] | None = None
def build_chat_params(
self,
default_template: str | None,
default_template_content_format: ChatTemplateContentFormatOption,
) -> ChatParams:
return ChatParams(
chat_template=default_template,
chat_template_content_format=default_template_content_format,
chat_template_kwargs=merge_kwargs(
merge_kwargs({"enable_thinking": False}, self.chat_template_kwargs),
dict(add_generation_prompt=True, continue_final_message=False),
),
)
def build_tok_params(self, model_config: ModelConfig) -> TokenizeParams:
return TokenizeParams(
max_total_tokens=model_config.max_model_len,
max_output_tokens=1,
add_special_tokens=False,
)
@@ -0,0 +1,160 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Question types for structured decisions.
A question type turns a request's criteria into options, writes the question
for the model with a label per option, and builds the answer from the label
probabilities.
"""
import math
import string
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, ClassVar
class StructuredDecisionError(ValueError):
"""An invalid request. Returned as a 400."""
#: Options are labeled A to Z in order. A single capital letter is one token
#: at the start of a reply for the tokenizers tested; every read checks it.
LABELS = tuple(string.ascii_uppercase)
@dataclass(frozen=True)
class Option:
name: str
description: str | None = None
@dataclass(frozen=True)
class Question:
id: str
type: "QuestionType"
instructions: str
options: tuple[Option, ...]
labels: tuple[str, ...]
class QuestionType(ABC):
name: ClassVar[str]
@abstractmethod
def parse_options(self, qid: str, criteria: Any) -> list[Option]: ...
@abstractmethod
def answer(
self, question: Question, probs: list[float], label_mass: float
) -> dict[str, Any]:
"""The answer for ``question``. ``probs[i]`` is the probability of
``question.labels[i]`` among the labels, and the list sums to 1.
``label_mass`` is the labels' total probability over the vocabulary."""
def prompt(self, question: Question) -> str:
"""The question as the model reads it, after the state."""
lines = [f"Question: {question.instructions}"] if question.instructions else []
for label, o in zip(question.labels, question.options):
lines.append(
f"{label}: {o.name} - {o.description}"
if o.description
else f"{label}: {o.name}"
)
lines.append("Answer with the letter of one option only.")
return "\n".join(lines)
QUESTION_TYPES: dict[str, QuestionType] = {}
def register_question_type(cls: type[QuestionType]) -> type[QuestionType]:
if cls.name in QUESTION_TYPES:
raise ValueError(f"question type {cls.name!r} is already registered")
QUESTION_TYPES[cls.name] = cls()
return cls
def get_question_type(name: str) -> QuestionType:
try:
return QUESTION_TYPES[name]
except KeyError:
raise StructuredDecisionError(
f"unknown question type {name!r}; supported: {sorted(QUESTION_TYPES)}"
) from None
def build_question(
qid: str,
type_name: str,
instructions: Any,
criteria: Any,
max_options: int,
) -> Question:
if not qid:
raise StructuredDecisionError("question ids must be non-empty")
qtype = get_question_type(type_name)
options = qtype.parse_options(qid, criteria)
if not options:
raise StructuredDecisionError(f"question {qid!r}: needs at least one option")
names = [o.name for o in options]
if len(set(names)) != len(names):
raise StructuredDecisionError(f"question {qid!r}: duplicate option names")
limit = min(max_options, len(LABELS))
if len(options) > limit:
raise StructuredDecisionError(
f"question {qid!r}: at most {limit} options for this model"
)
if not isinstance(instructions, str):
instructions = "" if instructions is None else str(instructions)
return Question(
id=qid,
type=qtype,
instructions=instructions,
options=tuple(options),
labels=LABELS[: len(options)],
)
def argmax(values: list[float]) -> int:
return max(range(len(values)), key=values.__getitem__)
def label_softmax(logprobs: list[float]) -> list[float]:
"""Softmax of the label logprobs."""
top = max(logprobs)
weights = [math.exp(lp - top) for lp in logprobs]
total = sum(weights)
return [w / total for w in weights]
@register_question_type
class ChoiceQuestion(QuestionType):
"""Pick one option. ``criteria`` maps each option name to a description,
or to null."""
name = "choice"
def parse_options(self, qid: str, criteria: Any) -> list[Option]:
if not isinstance(criteria, dict) or not criteria:
raise StructuredDecisionError(
f"question {qid!r}: choice criteria must map option names to "
"a description or null"
)
return [
Option(str(name), None if desc is None else str(desc))
for name, desc in criteria.items()
]
def answer(
self, question: Question, probs: list[float], label_mass: float
) -> dict[str, Any]:
top = argmax(probs)
return {
"type": self.name,
"choice": question.options[top].name,
"probabilities": {a.name: p for a, p in zip(question.options, probs)},
# The chosen label's probability over the whole vocabulary, so a
# read that mostly wanted a non-label token reports low confidence.
"confidence": probs[top] * label_mass,
}
@@ -0,0 +1,121 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import asyncio
import json
from typing import Any
from fastapi import Request
from vllm.entrypoints.openai.models.serving import OpenAIServingModels
from vllm.entrypoints.serve.engine.protocol import ErrorResponse
from vllm.entrypoints.serve.engine.serving import BaseServing
from vllm.entrypoints.serve.utils.request_logger import RequestLogger
from vllm.logger import init_logger
from .protocol import (
DecisionUsage,
QuestionDiagnostics,
StructuredDecisionRequest,
StructuredDecisionResponse,
)
from .question_types import Question, StructuredDecisionError, build_question
from .strategies import DecisionLimits, ReadStrategy
logger = init_logger(__name__)
def state_text(state: Any) -> str:
return state if isinstance(state, str) else json.dumps(state)
def parse_questions(
request: StructuredDecisionRequest,
limits: DecisionLimits,
) -> list[Question]:
if not request.questions:
raise StructuredDecisionError("questions: needs at least one question")
if len(request.questions) > limits.max_questions:
raise StructuredDecisionError(
f"questions: at most {limits.max_questions} for this model"
)
questions = []
for qid, spec in request.questions.items():
if spec.model_extra:
raise StructuredDecisionError(
f"question {qid!r}: unknown field(s) {sorted(spec.model_extra)}"
)
questions.append(
build_question(
qid,
spec.type,
spec.instructions,
spec.criteria,
limits.max_options,
)
)
return questions
class ServingStructuredDecisions(BaseServing):
def __init__(
self,
models: OpenAIServingModels,
strategy: ReadStrategy,
*,
request_logger: RequestLogger | None = None,
) -> None:
model_config = strategy.context.engine_client.model_config
super().__init__(
models=models, model_config=model_config, request_logger=request_logger
)
self.strategy = strategy
self.limits = strategy.limits()
async def create_decision(
self,
request: StructuredDecisionRequest,
raw_request: Request | None = None,
) -> StructuredDecisionResponse | ErrorResponse:
if (error := await self._check_model(request)) is not None: # type: ignore[arg-type]
return error
engine_client = self.strategy.context.engine_client
if engine_client.errored:
raise engine_client.dead_error
base_id = self._base_request_id(raw_request, default=request.request_id)
request_id = f"decision-{base_id}"
try:
questions = parse_questions(request, self.limits)
lora_request = self._maybe_get_adapters(request) # type: ignore[arg-type]
engine_client.check_admission(len(questions))
reads = await self.strategy.read(
questions,
request.instructions,
state_text(request.state),
request_id=request_id,
chat_template_kwargs=request.chat_template_kwargs,
lora_request=lora_request,
priority=request.priority,
)
except StructuredDecisionError as e:
return self.create_error_response(e)
except asyncio.CancelledError:
return self.create_error_response("Client disconnected")
answers: dict[str, dict[str, Any]] = {}
diagnostics: dict[str, QuestionDiagnostics] = {}
for q, read in zip(questions, reads):
answers[q.id] = q.type.answer(q, read.probs, read.label_mass)
diagnostics[q.id] = QuestionDiagnostics(
label_mass=read.label_mass, argmax_is_label=read.argmax_is_label
)
return StructuredDecisionResponse(
id=request_id,
model=self.models.model_name(lora_request),
answers=answers,
usage=DecisionUsage(
input_tokens=sum(r.input_tokens for r in reads),
output_tokens=sum(r.output_tokens for r in reads),
),
diagnostics=diagnostics,
)
@@ -0,0 +1,235 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Read strategies: how a model is asked for label probabilities.
Only models in NEXT_TOKEN_ARCHITECTURES are currently supported.
"""
import math
from abc import ABC, abstractmethod
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
from vllm.config import ModelConfig
from vllm.engine.protocol import EngineClient
from vllm.entrypoints.chat_utils import ChatTemplateContentFormatOption
from vllm.entrypoints.generate.label_reads import next_token_label_reads
from vllm.inputs import EngineInput
from vllm.lora.request import LoRARequest
from vllm.renderers.inputs.preprocess import extract_prompt_components
from vllm.renderers.online_renderer import OnlineRenderer
from vllm.sampling_params import SamplingParams
from vllm.tokenizers import TokenizerLike
from .protocol import ReadPromptRequest
from .question_types import LABELS, Question, StructuredDecisionError, label_softmax
@dataclass
class ReadContext:
engine_client: EngineClient
online_renderer: OnlineRenderer
chat_template: str | None
chat_template_content_format: ChatTemplateContentFormatOption
default_chat_template_kwargs: dict[str, Any]
@dataclass(frozen=True)
class DecisionLimits:
max_questions: int
max_options: int
@dataclass
class QuestionRead:
probs: list[float]
label_mass: float
argmax_is_label: bool
input_tokens: int
output_tokens: int
class ReadStrategy(ABC):
def __init__(self, context: ReadContext):
self.context = context
@abstractmethod
def limits(self) -> DecisionLimits: ...
@abstractmethod
async def read(
self,
questions: list[Question],
instructions: str | None,
state: str,
*,
request_id: str,
chat_template_kwargs: dict[str, Any] | None,
lora_request: LoRARequest | None,
priority: int,
) -> list[QuestionRead]:
"""One read per question, in order."""
def reply_label_ids(
tokenizer: TokenizerLike, prompt_ids: Sequence[int]
) -> tuple[list[int], list[int]]:
"""The prompt's ids after its last added token, and the token each of LABELS
adds after them as the first token of the reply. Raises ValueError if a
label is not one distinct token there."""
added = set(tokenizer.get_added_vocab().values())
start = max((i + 1 for i, t in enumerate(prompt_ids) if t in added), default=0)
tail = list(prompt_ids[start:])
text = tokenizer.decode(tail)
ids: list[int] = []
for label in LABELS:
extended = tokenizer.encode(text + label, add_special_tokens=False)
if extended[:-1] != tail or extended[-1] in ids:
raise ValueError(
f"label {label!r} is not one distinct token after this model's "
"chat prompt"
)
ids.append(extended[-1])
return tail, ids
class NextTokenStrategy(ReadStrategy):
"""Autoregressive models. Each question is one request: the state, then the
question with its labeled options, and the logprobs of the label tokens as
the reply's first token. The requests share the state. With prefix caching,
it is prefilled once."""
def __init__(self, context: ReadContext):
super().__init__(context)
tokenizer = context.online_renderer.renderer.get_tokenizer()
probe = tokenizer.apply_chat_template(
[{"role": "user", "content": "x"}],
**{
**context.default_chat_template_kwargs,
"chat_template": context.chat_template,
"add_generation_prompt": True,
"tokenize": True,
"return_dict": False,
"enable_thinking": False,
},
)
if isinstance(probe, str):
probe = tokenizer.encode(probe, add_special_tokens=False)
# Every prompt ends with the same generation prompt, so the labels'
# tokens are the same for every question.
self.tail, self.label_ids = reply_label_ids(tokenizer, probe)
def limits(self) -> DecisionLimits:
return DecisionLimits(max_questions=64, max_options=len(LABELS))
def _read_request(
self, chat_template_kwargs: dict[str, Any] | None
) -> ReadPromptRequest:
if (chat_template_kwargs or {}).get("enable_thinking"):
raise StructuredDecisionError(
"a read is the reply's first token, so thinking must be off"
)
return ReadPromptRequest(chat_template_kwargs=chat_template_kwargs)
async def _render(
self, read_request: ReadPromptRequest, messages: list[dict[str, Any]]
) -> tuple[EngineInput, list[int]]:
ctx = self.context
_, (engine_input,) = await ctx.online_renderer.preprocess_chat(
read_request,
messages,
default_template=ctx.chat_template,
default_template_content_format=ctx.chat_template_content_format,
default_template_kwargs=ctx.default_chat_template_kwargs,
)
prompt_ids = extract_prompt_components(
ctx.engine_client.model_config, engine_input
).token_ids
return engine_input, list(prompt_ids or [])
async def read(
self,
questions: list[Question],
instructions: str | None,
state: str,
*,
request_id: str,
chat_template_kwargs: dict[str, Any] | None,
lora_request: LoRARequest | None,
priority: int,
) -> list[QuestionRead]:
ctx = self.context
read_request = self._read_request(chat_template_kwargs)
slots, engine_inputs = [], []
for q in questions:
messages = [{"role": "user", "content": f"{state}\n\n{q.type.prompt(q)}"}]
if instructions:
messages.insert(0, {"role": "system", "content": instructions})
engine_input, prompt_ids = await self._render(read_request, messages)
if prompt_ids[len(prompt_ids) - len(self.tail) :] != self.tail:
raise StructuredDecisionError(
"these chat options end the prompt differently, so the "
"labels' tokens are unknown"
)
slots.append(self.label_ids[: len(q.labels)])
engine_inputs.append(engine_input)
label_reads = await next_token_label_reads(
ctx.engine_client,
engine_inputs,
[
SamplingParams(max_tokens=1, temperature=0.0, logprob_token_ids=ids)
for ids in slots
],
request_id,
lora_request=lora_request,
priority=priority,
)
reads = []
for ids, label_read in zip(slots, label_reads):
output = label_read.result.outputs[0]
reads.append(
QuestionRead(
probs=label_softmax(label_read.logprobs),
label_mass=sum(math.exp(lp) for lp in label_read.logprobs),
argmax_is_label=bool(output.token_ids)
and output.token_ids[0] in ids,
input_tokens=len(label_read.result.prompt_token_ids or ()),
output_tokens=len(output.token_ids),
)
)
return reads
NEXT_TOKEN_ARCHITECTURES = frozenset(
{
"Qwen3ForCausalLM",
"Qwen3_5ForConditionalGeneration",
"Qwen3_5MoeForConditionalGeneration",
}
)
# label_mass sums the labels' full-vocabulary probabilities. The logprobs modes
# return those probabilities. The logits modes return raw logits instead.
LOGPROBS_MODES = frozenset({"raw_logprobs", "processed_logprobs"})
def select_read_strategy(model_config: ModelConfig) -> type[ReadStrategy]:
"""Raises ValueError when the model cannot serve structured decisions. The
server opted in with --enable-structured-decisions, so startup fails."""
if model_config.architecture not in NEXT_TOKEN_ARCHITECTURES:
raise ValueError(
"--enable-structured-decisions does not support "
f"{model_config.architecture}. Supported architectures: "
f"{sorted(NEXT_TOKEN_ARCHITECTURES)}"
)
if model_config.logprobs_mode not in LOGPROBS_MODES:
raise ValueError(
"--enable-structured-decisions needs --logprobs-mode raw_logprobs or "
f"processed_logprobs, not {model_config.logprobs_mode}"
)
return NextTokenStrategy
+2
View File
@@ -163,6 +163,8 @@ class BaseFrontendArgs:
enable_tokenizer_info_endpoint: bool = False
"""Enable the `/tokenizer_info` endpoint. May expose chat
templates and other tokenizer configuration."""
enable_structured_decisions: bool = False
"""Enable the `/v1/systemone` structured decisions endpoint."""
enable_log_outputs: bool = False
"""If set to True, log model outputs (generations). Requires
`--enable-log-requests`. Output text and finish reasons are logged at INFO,