diff --git a/docs/serving/online_serving/README.md b/docs/serving/online_serving/README.md index 8e1a4f3398..fc46a482c1 100644 --- a/docs/serving/online_serving/README.md +++ b/docs/serving/online_serving/README.md @@ -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 diff --git a/docs/serving/online_serving/structured_decisions.md b/docs/serving/online_serving/structured_decisions.md new file mode 100644 index 0000000000..dd26f21301 --- /dev/null +++ b/docs/serving/online_serving/structured_decisions.md @@ -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. diff --git a/docs/usage/security.md b/docs/usage/security.md index 3dad19d6cb..c0d2c79700 100644 --- a/docs/usage/security.md +++ b/docs/usage/security.md @@ -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`) diff --git a/tests/entrypoints/generate/structured_decisions/__init__.py b/tests/entrypoints/generate/structured_decisions/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/entrypoints/generate/structured_decisions/test_question_types.py b/tests/entrypoints/generate/structured_decisions/test_question_types.py new file mode 100644 index 0000000000..e5906accf0 --- /dev/null +++ b/tests/entrypoints/generate/structured_decisions/test_question_types.py @@ -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"] diff --git a/tests/entrypoints/generate/structured_decisions/test_strategies.py b/tests/entrypoints/generate/structured_decisions/test_strategies.py new file mode 100644 index 0000000000..c4f4bce61b --- /dev/null +++ b/tests/entrypoints/generate/structured_decisions/test_strategies.py @@ -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:")) diff --git a/tests/entrypoints/generate/structured_decisions/test_structured_decisions_e2e.py b/tests/entrypoints/generate/structured_decisions/test_structured_decisions_e2e.py new file mode 100644 index 0000000000..e060b096ec --- /dev/null +++ b/tests/entrypoints/generate/structured_decisions/test_structured_decisions_e2e.py @@ -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"] diff --git a/vllm/entrypoints/generate/api_router.py b/vllm/entrypoints/generate/api_router.py index c5703515d8..70869b61b5 100644 --- a/vllm/entrypoints/generate/api_router.py +++ b/vllm/entrypoints/generate/api_router.py @@ -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 + ) diff --git a/vllm/entrypoints/generate/generative_scoring/serving.py b/vllm/entrypoints/generate/generative_scoring/serving.py index 4d5a15c2f5..bd46158edc 100644 --- a/vllm/entrypoints/generate/generative_scoring/serving.py +++ b/vllm/entrypoints/generate/generative_scoring/serving.py @@ -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) diff --git a/vllm/entrypoints/generate/label_reads.py b/vllm/entrypoints/generate/label_reads.py new file mode 100644 index 0000000000..eda61b22bc --- /dev/null +++ b/vllm/entrypoints/generate/label_reads.py @@ -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 diff --git a/vllm/entrypoints/generate/structured_decisions/__init__.py b/vllm/entrypoints/generate/structured_decisions/__init__.py new file mode 100644 index 0000000000..208f01a7cb --- /dev/null +++ b/vllm/entrypoints/generate/structured_decisions/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project diff --git a/vllm/entrypoints/generate/structured_decisions/api_router.py b/vllm/entrypoints/generate/structured_decisions/api_router.py new file mode 100644 index 0000000000..a0fbaa518e --- /dev/null +++ b/vllm/entrypoints/generate/structured_decisions/api_router.py @@ -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) diff --git a/vllm/entrypoints/generate/structured_decisions/protocol.py b/vllm/entrypoints/generate/structured_decisions/protocol.py new file mode 100644 index 0000000000..3a193e203b --- /dev/null +++ b/vllm/entrypoints/generate/structured_decisions/protocol.py @@ -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, + ) diff --git a/vllm/entrypoints/generate/structured_decisions/question_types.py b/vllm/entrypoints/generate/structured_decisions/question_types.py new file mode 100644 index 0000000000..a9a6514a03 --- /dev/null +++ b/vllm/entrypoints/generate/structured_decisions/question_types.py @@ -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, + } diff --git a/vllm/entrypoints/generate/structured_decisions/serving.py b/vllm/entrypoints/generate/structured_decisions/serving.py new file mode 100644 index 0000000000..7e2136b8a2 --- /dev/null +++ b/vllm/entrypoints/generate/structured_decisions/serving.py @@ -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, + ) diff --git a/vllm/entrypoints/generate/structured_decisions/strategies.py b/vllm/entrypoints/generate/structured_decisions/strategies.py new file mode 100644 index 0000000000..22233ff91b --- /dev/null +++ b/vllm/entrypoints/generate/structured_decisions/strategies.py @@ -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 diff --git a/vllm/entrypoints/launchers/cli_args.py b/vllm/entrypoints/launchers/cli_args.py index 28fe79fd69..18f8cb00b5 100644 --- a/vllm/entrypoints/launchers/cli_args.py +++ b/vllm/entrypoints/launchers/cli_args.py @@ -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,