From b05412a38ebfe21bb9f69184510a26bab0012f6a Mon Sep 17 00:00:00 2001 From: Valeriy Selitskiy <239034+iamwavecut@users.noreply.github.com> Date: Thu, 22 Oct 2026 02:30:25 +0200 Subject: [PATCH] feat(serve): typed decisions through /v1/systemone and /v1/decisions TypeSafe's System One API (as llama.cpp, vLLM and Ollama serve it) and OpenAI's Decisions API, answered by the served chat model: one read per question, the first reply token's exact log probabilities of the labels (A.. for a choice, yes/Yes and no/No for a noul, 0.. for a score) read by a new readout_logprobs Op on the head's logits in the Qwen3.5 core (ExecutionOptions::readout_tokens, TokenLogprob::readout); the other cores read them from the top-20 alternatives. A system turn keeps the first token on the labels (98-99% of its probability on Bonsai 2 27B), the reads run a lane's worth at a time and share the state's cached prefix, and the responses carry label_mass, argmax_is_label and readout diagnostics. --- README.md | 5 + docs/README.md | 2 +- docs/serving.md | 68 ++ include/ninfer/ops/readout_logprobs.h | 55 ++ include/ninfer/types.h | 13 + scripts/pods/decision_bench.py | 330 +++++++++ scripts/pods/decision_checks.sh | 12 + scripts/pods/decision_models.sh | 9 + src/models/qwen3_5/execution/text.cpp | 6 + src/models/qwen3_5/execution/text.h | 5 + .../qwen3_5/frontend/output_session.cpp | 4 + src/models/qwen3_5/program/context.h | 3 + src/models/qwen3_5/program/decode.cpp | 28 + .../qwen3_5/program/planning/request_plan.cpp | 7 + src/models/qwen3_5/program/prefill.cpp | 21 +- .../qwen3_5/program/prefix/hybrid_program.cpp | 2 + src/models/qwen3_5/program/program_impl.h | 14 + src/models/qwen3_5/program/round_buffers.cpp | 5 + src/models/qwen3_5/program/round_buffers.h | 6 + src/ops/basic_sources.cmake | 2 + src/ops/kernel/readout_logprobs.cuh | 52 ++ src/ops/launcher/readout_logprobs.cu | 20 + src/ops/launcher/readout_logprobs.h | 16 + src/ops/wrapper/readout_logprobs.cpp | 82 +++ src/runtime/contract/execution.h | 4 + src/runtime/contract/request.h | 3 + src/runtime/engine/engine.cpp | 8 + src/serve/CMakeLists.txt | 2 + src/serve/decisions_http.cpp | 112 +++ src/serve/http_server.cpp | 10 + src/serve/http_server.h | 4 + src/serve/openai_common.cpp | 4 + src/serve/openai_common.h | 1 + src/serve/request.h | 3 + src/serve/systemone.cpp | 670 ++++++++++++++++++ src/serve/systemone.h | 112 +++ src/serve/translate.cpp | 3 +- tests/cmake/ProductTests.cmake | 4 + tests/ops/test_readout_logprobs.cpp | 228 ++++++ tests/ops/tests.cmake | 1 + tests/test_systemone.cpp | 400 +++++++++++ 41 files changed, 2332 insertions(+), 4 deletions(-) create mode 100644 include/ninfer/ops/readout_logprobs.h create mode 100644 scripts/pods/decision_bench.py create mode 100644 scripts/pods/decision_checks.sh create mode 100644 scripts/pods/decision_models.sh create mode 100644 src/ops/kernel/readout_logprobs.cuh create mode 100644 src/ops/launcher/readout_logprobs.cu create mode 100644 src/ops/launcher/readout_logprobs.h create mode 100644 src/ops/wrapper/readout_logprobs.cpp create mode 100644 src/serve/decisions_http.cpp create mode 100644 src/serve/systemone.cpp create mode 100644 src/serve/systemone.h create mode 100644 tests/ops/test_readout_logprobs.cpp create mode 100644 tests/test_systemone.cpp diff --git a/README.md b/README.md index 6a4c0070d1..0270b3b8d6 100644 --- a/README.md +++ b/README.md @@ -309,6 +309,11 @@ Prefill of a 4,463-token prompt, in tok/s: `/tokenize`, `/detokenize`, `/apply-template`, `/v1/rerank`, `/metrics`, `/slots` and `/props`, streaming, tool calls, structured output through xgrammar (`--structured-output`), token log probabilities and prompt grafts ([HTTP serving](docs/serving.md)). +- **Typed decisions.** `/v1/systemone` (TypeSafe's System One API, as llama.cpp, vLLM and Ollama + serve it) and `/v1/decisions` (OpenAI's Decisions API) answer choice, yes/no and score questions + about a state with the served chat model: one prompt per question, the answer read from the + first token's exact label log probabilities, the state prefilled once + ([typed decisions](docs/serving.md#typed-decisions)). - **Model suspend.** `--model-suspend` lets an idle server give its device memory back without exiting and take it again on the next request, retained conversations included. Ternary Bonsai 2 27B on an RTX 3090: 7.9 GiB down to 0.3 GiB in 0.33 s, back in 1.1 s diff --git a/docs/README.md b/docs/README.md index 3b01bd6ea6..afbcdf442d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -11,7 +11,7 @@ spelling and defaults. | [Linux build](build-linux.md) | the container image, native Ubuntu builds for every target (`70`, `75`, `80`, `86`, `89`, `120a`), build options and the Bash launchers | | [Windows build](build-windows.md) | native Windows builds, their toolchain constraints and the launcher scripts | | [CLI](cli.md) | text, chat-history, image/video input, output streams, sampling, speculative decoding, and the runtime options | -| [HTTP serving](serving.md) | OpenAI Responses/Chat Completions, Anthropic Messages, llama.cpp endpoints, the router, model suspend, state, streaming, token counting, authentication, tool calls, and the server options | +| [HTTP serving](serving.md) | OpenAI Responses/Chat Completions, Anthropic Messages, llama.cpp endpoints, rerank and typed decisions, the router, model suspend, state, streaming, token counting, authentication, tool calls, and the server options | | [Ngram copy proposals](ngram.md) | copy acceleration alongside MTP, DFlash or DFlash2, on by default with a drafter | | [Launch recipes](recipes.md) | the `ninfer-serve` commands for every model and memory size, from an 8 GB card to the largest context, and the flags they use | | [Supported GPUs](gpus.md) | architectures and their builds, every card of 8 GB and more with what has run on it, and what fits in a given amount of GPU memory | diff --git a/docs/serving.md b/docs/serving.md index 4e099323e9..e6b5fc2b7e 100644 --- a/docs/serving.md +++ b/docs/serving.md @@ -461,6 +461,8 @@ up. A router instead unloads the failed model once its requests are gone and rep | `POST /tokenize`, `POST /detokenize` | the model's tokenizer, in llama.cpp's shape | | `POST /apply-template` | the prompt a chat request renders to, as text, without generation | | `POST /v1/rerank`, `POST /rerank` (and `/reranking`) | documents ranked by relevance to a query, scored by the model (see [Rerank](#rerank)) | +| `POST /v1/systemone` | typed questions about a state answered by the model, TypeSafe's System One API (see [Typed decisions](#typed-decisions)) | +| `POST /v1/decisions` | the same reads in OpenAI's Decisions API shape | | `GET /models`, `POST /models/load\|unload\|sleep`, `GET /models/sse` | llama.cpp's router API: the catalog and its lifecycle (see [Several models](#several-models-router)) | | `POST /v1/responses` | OpenAI Responses Core generation, state, typed Items, and SSE | | `POST /v1/responses/input_tokens` | Responses prompt-token count without generation | @@ -1379,6 +1381,72 @@ array of `{"index", "score"}`, with `text` when `return_text` is set. The judgem worth (`--max-concurrency`) at a time and share the cached judging prefix; on an RTX 3090 with a 27B model and two lanes, four short documents took 0.4 s. +## Typed decisions + +`POST /v1/systemone` answers typed questions about a `state` in the shape of TypeSafe's System One +API, as llama.cpp, vLLM and Ollama serve it, and `POST /v1/decisions` takes the same reads in the +shape of OpenAI's Decisions API. The served chat model is the decision model: each question becomes +one prompt -- a fixed system turn asking for one label, then the state, the question and its +options under one-token labels (`A`.. for a choice, `yes`/`no` for a noul, `0`.. for a score), +thinking off -- and its answer is read from the first reply token's exact log probabilities of +those labels (a noul's `yes` and `Yes`, `no` and `No` added up), without sampling past it. No +startup option is needed; a Qwen3.5-core model reads the labels exactly, Flash-Next and Gemma 4 +read them from the token's 20 most likely alternatives (`diagnostics..readout` says which). + +```bash +curl http://127.0.0.1:8080/v1/systemone -d '{ + "state": {"ticket": "My card was charged twice for one order."}, + "questions": { + "team": {"type": "choice", "instructions": "Which team should handle this?", + "criteria": {"billing": "payments and refunds", "shipping": "deliveries"}}, + "urgent": {"type": "noul", "instructions": "Does the customer need a reply within the hour?"}, + "tone": {"type": "score", "instructions": "How upset is the customer?", + "criteria": ["calm", "annoyed", "furious"]}}}' +``` + +The request takes `state` (a string, or JSON that the prompt shows as its compact text), +`questions` (an object of 1 to 64 question ids, each `{type, instructions, criteria}`, asked in +request order) and optionally `instructions` (context placed ahead of the state) and +`chat_template_kwargs` (accepted for compatibility; thinking stays off). A `choice` lists 2 to 26 +options as `{name: description or null}`, a `score` 2 to 10 level descriptions lowest first, and a +`noul` takes no criteria or `{"true": ..., "false": ...}` descriptions. Unknown fields, `joint` +reads and a thinking request are refused with 400; `images`, `files`, `audio` and `videos` with 501. + +The response holds `answers` by question id: a `choice` with the most likely option, its +`probabilities` by option (summing to 1) and `confidence`; a `noul` with `noul` (the probability of +yes) and the same `probabilities` and `confidence`; a `score` with `score` (the expected level index, +weighted by probability), the `legend` and `probabilities` by level index, and `confidence`. +Probabilities are the softmax over the labels; `confidence` is the chosen label's probability over +the whole vocabulary, its share of the labels times `diagnostics..label_mass`, the probability +the model put on the labels at all. `diagnostics..argmax_is_label` says whether the model's most +likely token was one of the labels, and `readout` whether the labels were read exactly (`exact`) or +from the top alternatives (`top_k`, where a label outside them counts as likely as the least likely +alternative, its upper bound). `usage.input_tokens` sums the prompts of every read and +`output_tokens` counts one token per question. The probabilities are those of a general chat model +under this prompt, not calibrated to any data, and the head's logits are BF16, so two options the +model scores within one BF16 step (about 0.1 nat at these magnitudes) share one probability; a label +the tokenizer does not hold as one token is refused with 400. + +`POST /v1/decisions` takes OpenAI's request -- `input` as a string or user messages with string +content or `input_text` parts, `questions` as an array of `{type, name, instructions}` with +`choices: [{value, description}]` (string or boolean values, echoed with their type) for a `choice` +and `levels: [{label, description}]` for a `score`; a `predicate` is a noul -- within the same +limits, and answers with `answers` in question order: a `predicate` with its `probability`, a +`choice` with `choice`, `probabilities: [{value, probability}]` and `confidence`, a `score` with +`score`, `probabilities: [{value, label, probability}]` and `confidence`; `name` is echoed, null for +an unnamed question. `safety_identifier` is accepted and ignored; streaming, tools, images and +refusal scoring are not implemented. + +The reads of one request run a lane's worth (`--max-concurrency`) at a time, so a request leaves +room for other clients' generation between its windows, and every read starts with the state, which +the prefix cache prefills once: a question costs its own text and one token. On an RTX 3090 with +Ternary Bonsai 2 27B and four lanes, a request about a 370-token state took 17 ms for one question +and 235 ms for sixteen once the state was cached (150 ms and 0.8 s with the state's prefill); about +a 13.5K-token state, 39 ms and 0.33 s (5.3 s and 6.1 s cold). Sixteen questions beside a streaming +256-token chat completion took 0.42 s instead of 0.26 s and slowed the chat's decode from 90 to +83 tokens/s. With the system turn the labels held 98-99% of the first token's probability on that +model, where the user turn alone left 60-90%. + ## OpenAI Responses Core NInfer implements the typed-Item and semantic-event core of the OpenAI diff --git a/include/ninfer/ops/readout_logprobs.h b/include/ninfer/ops/readout_logprobs.h new file mode 100644 index 0000000000..4128eabd9a --- /dev/null +++ b/include/ninfer/ops/readout_logprobs.h @@ -0,0 +1,55 @@ +#pragma once + +#include "core/tensor.h" + +#include + +#include // cudaStream_t + +namespace ninfer::ops { + +// The most tokens one readout_logprobs() call reads. +inline constexpr std::int32_t kReadoutLogprobsMaximum = 32; + +/** + * Op: Readout token log-probabilities + * + * The log probabilities of a caller-chosen set of tokens under one logits column, the typed + * decision readout: a request names the tokens its answer can be and reads their probabilities + * at the first generated position, without sampling them. + * + * Math / indexing: + * Let l[r] be the exact real value represented by logits[r]. For every k in [0,count), + * + * ideal[k] = l[token_ids[k]] - log(sum_{r=0..valid_rows-1} exp(l[r])). + * + * Logical shapes: + * logits is [physical_rows] or [physical_rows,1], token_ids is [capacity] and output is + * [capacity], with 1<=count<=capacity<=kReadoutLogprobsMaximum and + * 1<=valid_rows<=physical_rows. The first `count` values of token_ids are in [0,valid_rows); + * the rest of token_ids and output are not read or written. Physical rows + * [valid_rows,physical_rows) take no part in the denominator. + * + * Supported domain: + * logits is contiguous finite BF16, token_ids is contiguous I32, and output is contiguous + * FP32. Storage has its dtype's natural alignment. + * + * Numeric: + * output is the FP32 numerical approximation of ideal. Reduction association and private + * accumulator precision are implementation choices; the independent oracle evaluates the full + * formula in FP64 from the represented BF16 inputs. + * + * Effects: + * Writes output[0..count) and preserves both inputs. Output must not overlap logits or + * token_ids. + * + * Workspace: + * None. + * + * Execution: + * Enqueues work on stream and owns no persistent state. + */ +void readout_logprobs(const Tensor& logits, const Tensor& token_ids, std::int32_t count, + std::int32_t valid_rows, Tensor& output, cudaStream_t stream); + +} // namespace ninfer::ops diff --git a/include/ninfer/types.h b/include/ninfer/types.h index 8cf70efe3e..3021b0151a 100644 --- a/include/ninfer/types.h +++ b/include/ninfer/types.h @@ -861,6 +861,9 @@ struct StructuredOutputOptions { std::vector choices; }; +// The most token ids one request may name for the typed readout (ExecutionOptions::readout_tokens). +inline constexpr std::size_t kMaximumReadoutTokens = 32; + struct ExecutionOptions { StructuredOutputOptions structured_output; SamplingOverrides sampling; @@ -875,6 +878,12 @@ struct ExecutionOptions { // kMaximumTokenLogprobs most likely tokens under the distribution the sampler draws from // (see GenerationResult::content_logprobs). Sampling is unaffected. bool logprobs = false; + // Typed readout, with logprobs: the exact log probability of each of these token ids at the + // first generated token, under the softmax of the head's logits (the distribution of a greedy + // request without penalties), whether or not the id is among the top alternatives + // (TokenLogprob::readout). At most kMaximumReadoutTokens ids, each within the model's public + // token domain. A core without the readout leaves TokenLogprob::readout empty. + std::vector readout_tokens; }; struct OutputOptions { @@ -1275,6 +1284,10 @@ struct TokenLogprob { std::array top_values{}; // Decoded byte string for each top-k alternative, aligned with top_ids. std::array top_bytes; + // The log probability of each ExecutionOptions::readout_tokens id, in the request's order, + // under the softmax of the head's logits at the first generated token; empty on every other + // token and when the request named no readout tokens or the core has no typed readout. + std::vector readout; }; struct OutputDelta { diff --git a/scripts/pods/decision_bench.py b/scripts/pods/decision_bench.py new file mode 100644 index 0000000000..b6d7f82ec6 --- /dev/null +++ b/scripts/pods/decision_bench.py @@ -0,0 +1,330 @@ +#!/usr/bin/env python3 +"""The typed decision endpoints against a served model. + +usage: decision_bench.py [--model PATH] [--server PATH] [--out DIR] [--lanes N] + +Starts ninfer-serve on the model, then: + 1. answers: /v1/systemone and /v1/decisions on one short state, every question type; + 2. agreement: a choice question's probabilities against /v1/chat/completions' top log + probabilities of the same prompt (greedy, no penalties, thinking off); + 3. latency: /v1/systemone with 1, 4 and 16 questions on a ~200-token and a ~10K-token state, + a cold pass (the state prefilled) and warm passes (the state cached), per request wall time; + 4. concurrency: 16 questions while a 256-token chat completion streams, and the chat's decode + rate with and without the decisions beside it. +Writes DIR/decisions.json and prints one line per measurement. Exits nonzero when an answer, +agreement or shape check fails. +""" +from __future__ import annotations + +import argparse +import json +import math +import os +import signal +import statistics +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.request +from pathlib import Path + +ROOT = Path("/workspace/ninfer-work") +PORT = int(os.environ.get("DECISION_BENCH_PORT", "18091")) +BASE = f"http://127.0.0.1:{PORT}" +RUNNING: list[subprocess.Popen] = [] + +SENTENCES = [ + "The customer wrote on Monday that the parcel arrived with a cracked screen.", + "Support offered a replacement, but the customer asked for a refund instead.", + "The order was paid by card and the invoice shows two identical charges.", + "The warehouse confirmed that the package left on time with the usual courier.", + "A follow-up message on Wednesday repeated the request and sounded impatient.", + "The account has three earlier orders and no previous complaints.", +] + + +def call(path: str, body: dict | None = None, timeout: float = 600) -> dict: + data = None if body is None else json.dumps(body).encode() + request = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.load(response) + + +def call_status(path: str, body: dict) -> tuple[int, dict]: + data = json.dumps(body).encode() + request = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(request, timeout=600) as response: + return response.status, json.load(response) + except urllib.error.HTTPError as error: + return error.code, json.loads(error.read() or b"{}") + + +def stop_servers(*_) -> None: + for process in RUNNING: + if process.poll() is None: + try: + os.killpg(process.pid, signal.SIGTERM) + process.wait(timeout=60) + except (ProcessLookupError, subprocess.TimeoutExpired): + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + RUNNING.clear() + + +def handle_signal(number, _frame) -> None: + stop_servers() + raise SystemExit(128 + number) + + +def start_server(server: Path, model: Path, lanes: int, log_path: Path) -> subprocess.Popen: + command = [str(server), str(model), "--host", "127.0.0.1", "--port", str(PORT), + "--device", "0", "--max-context", "16384", "--kv-capacity", "65536", + "--max-concurrency", str(lanes)] + process = subprocess.Popen(command, stdout=log_path.open("w"), stderr=subprocess.STDOUT, + start_new_session=True) + RUNNING.append(process) + started = time.time() + while True: + if process.poll() is not None: + raise RuntimeError(f"server exited {process.returncode}; see {log_path}") + try: + call("/health", timeout=5) + return process + except (urllib.error.URLError, TimeoutError, ConnectionError, OSError): + if time.time() - started > 900: + raise RuntimeError("server start timed out") from None + time.sleep(2) + + +def state_text(target_tokens: int) -> str: + # About 15 tokens per sentence. + lines = [] + while len(lines) * 15 < target_tokens: + lines.append(f"Note {len(lines) + 1}: {SENTENCES[len(lines) % len(SENTENCES)]}") + return "\n".join(lines) + + +def questions(count: int) -> dict: + kinds = [ + ("choice", {"type": "choice", "instructions": "Which team should handle this?", + "criteria": {"billing": "payments and refunds", "shipping": "deliveries", + "technical": "defects and repairs"}}), + ("noul", {"type": "noul", "instructions": "Does the customer ask for a refund?"}), + ("score", {"type": "score", "instructions": "How upset is the customer?", + "criteria": ["calm", "annoyed", "furious"]}), + ("noul2", {"type": "noul", "instructions": "Is the parcel reported as damaged?", + "criteria": {"true": "the customer reports physical damage"}}), + ] + out = {} + for index in range(count): + name, body = kinds[index % len(kinds)] + out[f"{name}_{index}"] = body + return out + + +def check(condition: bool, label: str, failures: list[str]) -> None: + if not condition: + failures.append(label) + print(f"CHECK FAILED: {label}", flush=True) + + +def answers_section(record: dict, failures: list[str]) -> None: + state = state_text(200) + body = {"state": state, "questions": questions(4)} + status, response = call_status("/v1/systemone", body) + record["systemone"] = {"status": status, "response": response} + print(json.dumps({"systemone": response}, ensure_ascii=False), flush=True) + check(status == 200, "systemone returns 200", failures) + if status == 200: + check(response["object"] == "structured_decision" and response["usage"]["output_tokens"] == 4, + "systemone shape and usage", failures) + for name, answer in response["answers"].items(): + diagnostics = response["diagnostics"][name] + check(diagnostics["readout"] == "exact", f"{name} read from the exact readout", failures) + check(0.0 < diagnostics["label_mass"] <= 1.0 + 1e-6, f"{name} label mass in (0, 1]", + failures) + total = sum(answer["probabilities"].values()) + check(abs(total - 1.0) < 1e-6, f"{name} probabilities sum to 1 ({total})", failures) + if answer["type"] == "choice": + best = max(answer["probabilities"], key=answer["probabilities"].get) + check(answer["choice"] == best, f"{name} choice is the most likely option", failures) + elif answer["type"] == "score": + expected = sum(float(k) * p for k, p in answer["probabilities"].items()) + check(abs(answer["score"] - expected) < 1e-6, f"{name} score is the expectation", + failures) + else: + check(abs(answer["noul"] - answer["probabilities"]["yes"]) < 1e-9, + f"{name} noul is P(yes)", failures) + openai = {"input": [{"role": "user", "content": [{"type": "input_text", "text": state}]}], + "questions": [ + {"type": "predicate", "name": "refund", "instructions": "Does the customer ask for a refund?"}, + {"type": "choice", "name": "team", "instructions": "Which team should handle this?", + "choices": [{"value": "billing", "description": "payments and refunds"}, + {"value": "shipping", "description": "deliveries"}, + {"value": "technical", "description": "defects and repairs"}]}, + {"type": "score", "instructions": "How upset is the customer?", + "levels": [{"label": "calm"}, {"label": "annoyed"}, {"label": "furious"}]}]} + status, response = call_status("/v1/decisions", openai) + record["decisions"] = {"status": status, "response": response} + print(json.dumps({"decisions": response}, ensure_ascii=False), flush=True) + check(status == 200, "decisions returns 200", failures) + if status == 200: + check(response["object"] == "decision" and len(response["answers"]) == 3 and + response["answers"][2]["name"] is None, "decisions shape", failures) + same = record["systemone"]["response"]["answers"]["choice_0"]["probabilities"] + theirs = {entry["value"]: entry["probability"] + for entry in response["answers"][1]["probabilities"]} + check(all(abs(same[k] - theirs[k]) < 1e-3 for k in same), + "the same choice reads the same through both endpoints", failures) + status, response = call_status("/v1/systemone", {"state": state, "questions": { + "q": {"type": "choice", "instructions": "i", "criteria": {"only": None}}}}) + check(status == 400, "a one-option choice is refused", failures) + status, response = call_status("/v1/systemone", {"state": state, "images": ["data:image/png;base64,AA=="], + "questions": {"q": {"type": "noul", "instructions": "i"}}}) + check(status == 501, "images are not implemented", failures) + + +def agreement_section(record: dict, failures: list[str]) -> None: + state = state_text(200) + question = {"type": "choice", "instructions": "Which team should handle this?", + "criteria": {"billing": "payments and refunds", "shipping": "deliveries", + "technical": "defects and repairs"}} + decision = call("/v1/systemone", {"state": state, "questions": {"team": question}}) + # The prompt make_decision_read renders for this question: its system turn and user turn. + system = ("You answer typed questions about a state with one label only: the letter of the best " + "option, yes or no, or the number of the level. No words before or after the label.") + text = (f"State:\n{state}\n\nQuestion: {question['instructions']}\nOptions:\n" + "A. billing - payments and refunds\nB. shipping - deliveries\n" + "C. technical - defects and repairs\nReply with a single letter (A, B or C) and nothing else.") + chat = call("/v1/chat/completions", { + "messages": [{"role": "system", "content": system}, {"role": "user", "content": text}], + "max_tokens": 1, "temperature": 0, + "presence_penalty": 0, "frequency_penalty": 0, "logprobs": True, "top_logprobs": 20, + "graft": "", "chat_template_kwargs": {"enable_thinking": False}}) + top = chat["choices"][0]["logprobs"]["content"][0]["top_logprobs"] + by_token = {entry["token"]: entry["logprob"] for entry in top} + labels = {"A": "billing", "B": "shipping", "C": "technical"} + chat_probabilities = {} + for label, option in labels.items(): + if label in by_token: + chat_probabilities[option] = math.exp(by_token[label]) + mass = sum(chat_probabilities.values()) + ours = decision["answers"]["team"]["probabilities"] + diagnostics = decision["diagnostics"]["team"] + record["agreement"] = {"decision": decision, "chat_top": top, + "chat_label_probabilities": chat_probabilities} + print(json.dumps({"agreement": {"ours": ours, "label_mass": diagnostics["label_mass"], + "chat_labels_in_top20": chat_probabilities, + "chat_label_mass": mass}}), flush=True) + check(len(chat_probabilities) >= 1, "the chat's top-20 shows at least one label", failures) + if chat_probabilities: + for option, probability in chat_probabilities.items(): + check(abs(probability - ours[option] * diagnostics["label_mass"]) < 2e-3, + f"{option}: readout {ours[option] * diagnostics['label_mass']:.5f} vs chat " + f"{probability:.5f}", failures) + check(abs(mass - diagnostics["label_mass"]) < 5e-3 or len(chat_probabilities) < 3, + "label mass agrees when every label is in the top 20", failures) + + +def timed(path: str, body: dict) -> tuple[float, dict]: + begun = time.time() + response = call(path, body) + return time.time() - begun, response + + +def latency_section(record: dict, lanes: int) -> None: + results = [] + for state_tokens in (200, 10000): + for count in (1, 4, 16): + state = state_text(state_tokens) + f"\nRequest marker {state_tokens}-{count}." + body = {"state": state, "questions": questions(count)} + cold, response = timed("/v1/systemone", body) + warm = [] + for _ in range(3): + wall, response = timed("/v1/systemone", body) + warm.append(wall) + masses = [d["label_mass"] for d in response["diagnostics"].values()] + argmax = sum(1 for d in response["diagnostics"].values() if d["argmax_is_label"]) + row = {"state_tokens": response["usage"]["input_tokens"] // count, "questions": count, + "cold_s": round(cold, 3), "warm_s": [round(w, 3) for w in warm], + "warm_median_s": round(statistics.median(warm), 3), + "per_question_ms": round(1000 * statistics.median(warm) / count, 1), + "label_mass_min": round(min(masses), 4), "label_mass_mean": round(statistics.mean(masses), 4), + "argmax_is_label": f"{argmax}/{count}", "lanes": lanes} + results.append(row) + print(json.dumps({"latency": row}), flush=True) + record["latency"] = results + + +def chat_stream(max_tokens: int, out: dict) -> None: + body = {"messages": [{"role": "user", "content": "Write a long story about a lighthouse keeper."}], + "max_tokens": max_tokens, "temperature": 0, "ignore_eos": True, + "chat_template_kwargs": {"enable_thinking": False}} + begun = time.time() + response = call("/v1/chat/completions", body) + out["wall"] = time.time() - begun + out["timings"] = response.get("timings", {}) + + +def concurrency_section(record: dict) -> None: + state = state_text(2000) + "\nConcurrency marker." + body = {"state": state, "questions": questions(16)} + call("/v1/systemone", body) # the state is cached + alone_chat: dict = {} + chat_stream(256, alone_chat) + alone_decision, _ = timed("/v1/systemone", body) + shared_chat: dict = {} + thread = threading.Thread(target=chat_stream, args=(256, shared_chat)) + thread.start() + time.sleep(0.5) + shared_decision, _ = timed("/v1/systemone", body) + thread.join() + row = {"decision_alone_s": round(alone_decision, 3), "decision_beside_chat_s": round(shared_decision, 3), + "chat_alone_tok_s": alone_chat["timings"].get("predicted_per_second"), + "chat_alone_wall_s": round(alone_chat["wall"], 3), + "chat_beside_decisions_tok_s": shared_chat["timings"].get("predicted_per_second"), + "chat_beside_decisions_wall_s": round(shared_chat["wall"], 3)} + record["concurrency"] = row + print(json.dumps({"concurrency": row}), flush=True) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--model", default=str(ROOT / "models/Ternary-Bonsai-2-27B-ninfer-v3.ninfer")) + parser.add_argument("--server", default=str(ROOT / "build/apps/ninfer-serve")) + parser.add_argument("--out", default=os.environ.get("NINFER_JOB_DIR", ".")) + parser.add_argument("--lanes", type=int, default=4) + arguments = parser.parse_args() + out = Path(arguments.out) + out.mkdir(parents=True, exist_ok=True) + for number in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP): + signal.signal(number, handle_signal) + record: dict = {"model": arguments.model, "lanes": arguments.lanes} + failures: list[str] = [] + try: + process = start_server(Path(arguments.server), Path(arguments.model), arguments.lanes, + out / "decisions.server.log") + record["server_pid"] = process.pid + answers_section(record, failures) + agreement_section(record, failures) + latency_section(record, arguments.lanes) + concurrency_section(record) + finally: + stop_servers() + (out / "decisions.json").write_text(json.dumps(record, ensure_ascii=False, indent=1)) + if failures: + print(f"DECISION_BENCH_FAIL {len(failures)} check(s)", flush=True) + return 1 + print("DECISION_BENCH_PASS", flush=True) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/pods/decision_checks.sh b/scripts/pods/decision_checks.sh new file mode 100644 index 0000000000..c89ab47e4d --- /dev/null +++ b/scripts/pods/decision_checks.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# The typed decision endpoints on a GPU: the readout Op and serve tests, then the served model's +# answers, their agreement with the chat endpoint's log probabilities, latency by state length +# and question count, and the endpoint beside a concurrent chat request. +set -Eeuo pipefail +root=/workspace/ninfer-work +build="$root/build" +ctest --test-dir "$build" --output-on-failure -j2 \ + -R 'ninfer_(readout_logprobs|target_logprobs|logprob_topk|systemone|rerank)_test' +while [ ! -f "$root/jobs/models/exit" ]; do sleep 2; done +[ "$(cat "$root/jobs/models/exit")" = 0 ] || exit 125 +"$root/py311/bin/python" scripts/pods/decision_bench.py "$@" diff --git a/scripts/pods/decision_models.sh b/scripts/pods/decision_models.sh new file mode 100644 index 0000000000..44eb03d2f6 --- /dev/null +++ b/scripts/pods/decision_models.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# The artifact the decision checks serve: the published Ternary Bonsai 2 27B, a Qwen3.5-core model. +set -Eeuo pipefail +root=/workspace/ninfer-work +mkdir -p "$root/models" +[ -f "$root/models/Ternary-Bonsai-2-27B-ninfer-v3.ninfer" ] || \ + "$root/py311/bin/hf" download WaveCut/Ternary-Bonsai-2-27B-NInfer-v3 \ + Ternary-Bonsai-2-27B-ninfer-v3.ninfer --local-dir "$root/models" +ls -l "$root/models/Ternary-Bonsai-2-27B-ninfer-v3.ninfer" diff --git a/src/models/qwen3_5/execution/text.cpp b/src/models/qwen3_5/execution/text.cpp index 6021a7e3bd..06c4ec0070 100644 --- a/src/models/qwen3_5/execution/text.cpp +++ b/src/models/qwen3_5/execution/text.cpp @@ -22,6 +22,7 @@ #include "ninfer/ops/gdn_gating_proj.h" #include "ninfer/ops/gdn_input_proj.h" #include "ninfer/ops/logprob_topk.h" +#include "ninfer/ops/readout_logprobs.h" #include "ninfer/ops/linear.h" #include "ninfer/ops/kv_cache_append.h" #include "ninfer/ops/speculative_round.h" @@ -1642,6 +1643,11 @@ TextContext::prefill_impl(std::span ids, const TextPrefill* text_pref dimension(parameters_.model.resources().public_token_count), ids, values, lse, io_.logprob_active, work_, s); } + if (readout_count_ > 0) { + ops::readout_logprobs(logits, io_.readout_ids, readout_count_, + dimension(parameters_.model.resources().public_token_count), + io_.readout_values, s); + } // Set io_.pos to the bonus token's absolute position (base + T) before picking so // the sampler RNG is keyed by it (prefill purpose keeps it distinct from the first // decode step, which reuses the same io_.pos). diff --git a/src/models/qwen3_5/execution/text.h b/src/models/qwen3_5/execution/text.h index 4c5b137dc2..f5991372f7 100644 --- a/src/models/qwen3_5/execution/text.h +++ b/src/models/qwen3_5/execution/text.h @@ -94,6 +94,10 @@ class TextContext { void set_sampling(const ops::SamplingConfig* config) noexcept { sampling_config_ = config; } + // The typed readout a finalizing chunk performs on its sampled token's logits: the first + // `count` ids of io.readout_ids, read into io.readout_values; zero reads nothing. + void set_readout(std::int32_t count) noexcept { readout_count_ = count; } + void set_rope_yarn(const ops::RopeYarn& yarn) noexcept { rope_yarn_ = yarn; } // Keys before its first query that the MTP draft attention reads past its one-page sink; @@ -301,6 +305,7 @@ class TextContext { const std::int32_t* proposal_head_ids_ = nullptr; int proposal_head_n_ = 0; const ops::SamplingConfig* sampling_config_ = nullptr; + std::int32_t readout_count_ = 0; ops::RopeYarn rope_yarn_; std::uint32_t mtp_attention_window_ = 0; const MtpParameters* mtp_ = nullptr; diff --git a/src/models/qwen3_5/frontend/output_session.cpp b/src/models/qwen3_5/frontend/output_session.cpp index 7f2c12f51a..e38cb86d2e 100644 --- a/src/models/qwen3_5/frontend/output_session.cpp +++ b/src/models/qwen3_5/frontend/output_session.cpp @@ -616,6 +616,10 @@ class OutputSession::Impl { record.top_bytes[k] = std::string(tokenizer->decoded_token(raw.top_ids[k]).bytes); } } + if (raw.readout_count > 0) { + const std::size_t count = std::min(raw.readout_count, kMaximumReadoutTokens); + record.readout.assign(raw.readout_values.begin(), raw.readout_values.begin() + count); + } if (!fed) { decoder.utf8_logprobs.push_back(std::move(record)); return; diff --git a/src/models/qwen3_5/program/context.h b/src/models/qwen3_5/program/context.h index 11a33ca2e9..442974836e 100644 --- a/src/models/qwen3_5/program/context.h +++ b/src/models/qwen3_5/program/context.h @@ -88,6 +88,9 @@ struct PrefillContext { // the landing batch, which the cache's next poll() frees, so each chunk function calls this // immediately before its pass and nothing outside the chunk call can hold the view. std::function()> take_layer_ready; + // The typed readout's token count: the first generated token's logits are read at + // io.readout_ids[0..readout_count) into io.readout_values; zero reads nothing. + std::int32_t readout_count = 0; }; struct OrdinaryBatchContext { diff --git a/src/models/qwen3_5/program/decode.cpp b/src/models/qwen3_5/program/decode.cpp index fc5e5f9c73..3cba4e1367 100644 --- a/src/models/qwen3_5/program/decode.cpp +++ b/src/models/qwen3_5/program/decode.cpp @@ -248,6 +248,34 @@ runtime::RawTokenLogprob ProgramImpl::fetched_logprob(std::int32_t row, TokenId return record; } +std::int32_t ProgramImpl::arm_readout(std::span ids) { + if (ids.empty()) { return 0; } + if (ids.size() > kMaximumReadoutTokens || + static_cast(ids.size()) > io.readout_ids.ne[0]) { + throw std::logic_error("readout token ids exceed the round buffer"); + } + // A pageable source: the copy completes on the host before it returns. + CUDA_CHECK(cudaMemcpyAsync(io.readout_ids.data, ids.data(), ids.size() * sizeof(TokenId), + cudaMemcpyHostToDevice, device.stream)); + return static_cast(ids.size()); +} + +void ProgramImpl::fetch_readout(std::int32_t count) { + if (count <= 0) { return; } + if (!readout_host) { readout_host.emplace(kMaximumReadoutTokens * sizeof(float)); } + CUDA_CHECK(cudaMemcpyAsync(readout_host->data(), io.readout_values.data, + static_cast(count) * sizeof(float), + cudaMemcpyDeviceToHost, device.stream)); +} + +void ProgramImpl::attach_readout(runtime::RawTokenLogprob& record, std::int32_t count) const { + record.readout_count = 0; + if (count <= 0) { return; } + const auto* values = static_cast(readout_host->data()); + std::copy_n(values, count, record.readout_values.begin()); + record.readout_count = static_cast(count); +} + void ProgramImpl::copy_tail(SequenceState& sequence, const Tensor& source) { if (source.dtype != DType::BF16 || source.ne[0] != dimension(parameters.model.config().text.hidden_size) || diff --git a/src/models/qwen3_5/program/planning/request_plan.cpp b/src/models/qwen3_5/program/planning/request_plan.cpp index 9811b10e74..94d40d254d 100644 --- a/src/models/qwen3_5/program/planning/request_plan.cpp +++ b/src/models/qwen3_5/program/planning/request_plan.cpp @@ -318,6 +318,12 @@ RequestBasePlan ProgramImpl::plan_request(const PreparedPromptData& prompt, base->sampling = translate_sampling(options.sampling); base->grammar = options.grammar; base->logprobs = options.logprobs; + base->readout_tokens = options.readout_tokens; + for (const TokenId id : base->readout_tokens) { + if (id < 0 || id >= execution::dimension(parameters.model.resources().public_token_count)) { + throw std::invalid_argument("readout token is outside the public token domain"); + } + } base->allow_prefix_reuse = options.allow_prefix_reuse; base->summary.publish_continuation = options.allow_prefix_reuse && prompt.identity.reusable && context_cache.enabled; @@ -548,6 +554,7 @@ std::optional ProgramImpl::inspect_lane( plan->sampling = base.sampling; plan->grammar = base.grammar; plan->logprobs = base.logprobs; + plan->readout_tokens = base.readout_tokens; plan->text_kv_page_entitlement = base.text_kv_page_entitlement; plan->backend_kv_page_entitlement = base.backend_kv_page_entitlement; plan->root_rebuild_work = base.root_rebuild_work; diff --git a/src/models/qwen3_5/program/prefill.cpp b/src/models/qwen3_5/program/prefill.cpp index 8786227488..442687d149 100644 --- a/src/models/qwen3_5/program/prefill.cpp +++ b/src/models/qwen3_5/program/prefill.cpp @@ -6,6 +6,7 @@ #include "core/device.h" #include "ninfer/ops/gdn_replay.h" #include "ninfer/ops/logprob_topk.h" +#include "ninfer/ops/readout_logprobs.h" #include "ninfer/ops/sampling.h" #include "ninfer/ops/scalar.h" #include "ninfer/ops/scatter.h" @@ -100,6 +101,7 @@ PrefillChunkResult prefill_text_chunk(PrefillContext& state, std::span(*split_frontier) : -1); @@ -177,6 +179,8 @@ void mtp_bridge_multimodal(PrefillContext& state, const PreparedPromptData& prom bridge.rope_position, false, composed_embedding); } +static_assert(kMaximumReadoutTokens == static_cast(ops::kReadoutLogprobsMaximum)); + void sample_from_hidden(PrefillContext& state, const Tensor& hidden, std::int32_t absolute_position, std::int32_t purpose) { if (hidden.dtype != DType::BF16 || @@ -190,6 +194,11 @@ void sample_from_hidden(PrefillContext& state, const Tensor& hidden, std::int32_ state.execution.device.stream); // Before sampling adds the token to the penalty counts. gather_logprobs(state.execution, logits.view({logits.ne[0], 1, 1}), state.sampling, nullptr); + if (state.readout_count > 0) { + ops::readout_logprobs(logits, state.execution.io.readout_ids, state.readout_count, + dimension(state.execution.parameters.model.resources().public_token_count), + state.execution.io.readout_values, state.execution.device.stream); + } CUDA_CHECK(cudaMemcpyAsync(state.execution.io.pos.data, &absolute_position, sizeof(absolute_position), cudaMemcpyHostToDevice, state.execution.device.stream)); @@ -718,6 +727,7 @@ void ProgramImpl::start_sequence(std::uint32_t lane, SequenceState& sequence, ensure_sequence_kv_mapped(sequence, prompt_tokens, backend_materialized); request.grammar = request_plan.grammar; request.logprobs = request_plan.logprobs; + request.readout_tokens = request_plan.readout_tokens; install_sampling(sequence, request, request_plan.sampling); sequence.rope_delta = staged.prompt.rope_delta; set_device_i32(io.rope_delta, sequence.rope_delta); @@ -1177,6 +1187,7 @@ runtime::PrefillStepResult ProgramImpl::advance_prefill(SequenceState& sequence, return hybrid_take_restore_layers(lane); }; arm_logprobs(request.logprobs); + schedule_state.readout_count = arm_readout(request.readout_tokens); if (staged.mtp_bridge == MtpBridgeMode::BeforeSuffix) { if (staged.cursor != staged.base || staged.base == 0 || @@ -1367,7 +1378,10 @@ runtime::PrefillStepResult ProgramImpl::advance_prefill(SequenceState& sequence, } copy_round_token(); - if (request.logprobs) { fetch_logprobs(1); } + if (request.logprobs) { + fetch_logprobs(1); + fetch_readout(schedule_state.readout_count); + } std::array initial_drafts{}; if (staged.prepare_mtp && staged.initial_mtp_extent != 0) { CUDA_CHECK(cudaMemcpyAsync(initial_drafts.data(), io.mtp->draft_tokens.data, @@ -1382,7 +1396,10 @@ runtime::PrefillStepResult ProgramImpl::advance_prefill(SequenceState& sequence, const std::uint32_t prompt_tokens = staged.prompt_tokens; validate_licensed_tokens(std::span(host_tokens, 1)); - if (request.logprobs) { round_logprobs[0] = fetched_logprob(0, host_tokens[0]); } + if (request.logprobs) { + round_logprobs[0] = fetched_logprob(0, host_tokens[0]); + attach_readout(round_logprobs[0], schedule_state.readout_count); + } if (sequence.ledger.size() != prompt_tokens) { throw std::logic_error("candidate token ledger does not match prompt length"); } diff --git a/src/models/qwen3_5/program/prefix/hybrid_program.cpp b/src/models/qwen3_5/program/prefix/hybrid_program.cpp index fc1ffbcfea..61d353abe2 100644 --- a/src/models/qwen3_5/program/prefix/hybrid_program.cpp +++ b/src/models/qwen3_5/program/prefix/hybrid_program.cpp @@ -223,6 +223,7 @@ HybridAdmissionQuote ProgramImpl::hybrid_quote(const PreparedPromptData& prompt, quote->sampling = plan.sampling; quote->grammar = plan.grammar; quote->logprobs = plan.logprobs; + quote->readout_tokens = plan.readout_tokens; quote->text_kv_page_entitlement = plan.text_kv_page_entitlement; quote->backend_kv_page_entitlement = plan.backend_kv_page_entitlement; quote->vision_control_plan = plan.vision_control_plan; @@ -918,6 +919,7 @@ StartResult ProgramImpl::hybrid_activate(HybridMaterializationTransaction& trans ensure_sequence_kv_mapped(sequence, n, backend_materialized); request.grammar = quote.grammar; request.logprobs = quote.logprobs; + request.readout_tokens = quote.readout_tokens; install_sampling(sequence, request, quote.sampling); sequence.mtp_draft_count = 0; sequence.rope_delta = prompt.rope_delta; diff --git a/src/models/qwen3_5/program/program_impl.h b/src/models/qwen3_5/program/program_impl.h index 9735541bb7..e68b8beb02 100644 --- a/src/models/qwen3_5/program/program_impl.h +++ b/src/models/qwen3_5/program/program_impl.h @@ -217,6 +217,8 @@ struct RequestBasePlanImpl { std::shared_ptr grammar; // The request asked for each generated token's log probability. bool logprobs = false; + // With logprobs: the token ids read out exactly on the first generated token. + std::vector readout_tokens; std::uint32_t text_kv_page_entitlement = 0; std::uint32_t backend_kv_page_entitlement = 0; std::shared_ptr vision_control_plan; @@ -283,6 +285,7 @@ struct AdmissionCandidateImpl : ResourceCandidateState { ops::SamplingConfig sampling; std::shared_ptr grammar; bool logprobs = false; + std::vector readout_tokens; std::uint32_t text_kv_page_entitlement = 0; std::uint32_t backend_kv_page_entitlement = 0; runtime::LaneId destination{}; @@ -305,6 +308,7 @@ struct HybridQuoteImpl { ops::SamplingConfig sampling; std::shared_ptr grammar; bool logprobs = false; + std::vector readout_tokens; std::uint32_t text_kv_page_entitlement = 0; std::uint32_t backend_kv_page_entitlement = 0; std::shared_ptr vision_control_plan; @@ -542,6 +546,8 @@ struct RequestControl { std::uint32_t next_mask_frontier = 0; std::chrono::steady_clock::time_point next_mask_started; bool logprobs = false; + // The typed readout's token ids; read on the sequence's first generated token only. + std::vector readout_tokens; GenerationTimings timings; SpeculativeStats speculative_stats; // Adaptive MTP: how far this request's drafts survive. @@ -987,6 +993,8 @@ class ProgramImpl { round_logprobs{}; std::optional logprobs_host; std::optional logprobs_armed; + // The host copy of a prefill's typed readout (io.readout_values). + std::optional readout_host; std::optional ordinary_host; qwen3_5::OrdinaryDecodeIngress* ordinary_host_ingress = nullptr; @@ -1693,6 +1701,12 @@ class ProgramImpl { void fetch_logprobs(std::int32_t rows); // The fetched record of gather row `row`, for the token that row produced. [[nodiscard]] runtime::RawTokenLogprob fetched_logprob(std::int32_t row, TokenId token) const; + // Uploads a request's readout token ids ahead of its prefill and returns their count. + std::int32_t arm_readout(std::span ids); + // Copies the first `count` readout values to the host, ordered after the enqueued prefill. + void fetch_readout(std::int32_t count); + // Attaches the fetched readout values to the first token's record. + void attach_readout(runtime::RawTokenLogprob& record, std::int32_t count) const; void copy_tail(SequenceState& sequence, const Tensor& source); void copy_round_token(); void diff --git a/src/models/qwen3_5/program/round_buffers.cpp b/src/models/qwen3_5/program/round_buffers.cpp index f00cc49feb..a775ce9575 100644 --- a/src/models/qwen3_5/program/round_buffers.cpp +++ b/src/models/qwen3_5/program/round_buffers.cpp @@ -103,6 +103,9 @@ RoundStateLayout begin_round_state_layout(LayoutBuilder& builder, const RoundSta add_tensor(builder, DType::FP32, {ops::kLogprobTopK, rows}, "logprob top values"); layout.logprob_lse = add_tensor(builder, DType::FP32, {rows}, "logprob lse"); layout.logprob_active = add_tensor(builder, DType::I32, {1}, "logprob gather flag"); + const auto readout = static_cast(kMaximumReadoutTokens); + layout.readout_ids = add_tensor(builder, DType::I32, {readout}, "readout token ids"); + layout.readout_values = add_tensor(builder, DType::FP32, {readout}, "readout logprobs"); } return layout; } @@ -483,6 +486,8 @@ RoundState::RoundState(DeviceSpan backing, const RoundStateLayout& layout) { logprob_values = layout.logprob_values.bind(backing); logprob_lse = layout.logprob_lse.bind(backing); logprob_active = layout.logprob_active.bind(backing); + readout_ids = layout.readout_ids.bind(backing); + readout_values = layout.readout_values.bind(backing); } if (layout.mtp) { mtp.emplace(backing, *layout.mtp); } if (layout.dflash_prefill) { dflash_prefill.emplace(backing, *layout.dflash_prefill); } diff --git a/src/models/qwen3_5/program/round_buffers.h b/src/models/qwen3_5/program/round_buffers.h index be9c1c84e5..3f7a4017bb 100644 --- a/src/models/qwen3_5/program/round_buffers.h +++ b/src/models/qwen3_5/program/round_buffers.h @@ -193,6 +193,8 @@ struct RoundStateLayout { TensorRegion logprob_values; TensorRegion logprob_lse; TensorRegion logprob_active; + TensorRegion readout_ids; + TensorRegion readout_values; std::optional mtp; std::optional dflash_prefill; std::optional mtp_decode; @@ -354,6 +356,10 @@ struct RoundState { Tensor logprob_values; Tensor logprob_lse; Tensor logprob_active; + // The typed readout of a prefill's first token: I32 [kMaximumReadoutTokens] ids a request + // named and the FP32 [kMaximumReadoutTokens] log probabilities read for them. + Tensor readout_ids; + Tensor readout_values; std::optional mtp; std::optional dflash_prefill; std::optional mtp_decode; diff --git a/src/ops/basic_sources.cmake b/src/ops/basic_sources.cmake index fc3b8a615c..0ee7b7545f 100644 --- a/src/ops/basic_sources.cmake +++ b/src/ops/basic_sources.cmake @@ -29,6 +29,7 @@ target_sources(ninfer_ops PRIVATE "${CMAKE_CURRENT_LIST_DIR}/launcher/scatter.cu" "${CMAKE_CURRENT_LIST_DIR}/launcher/sigmoid_gate_mul.cu" "${CMAKE_CURRENT_LIST_DIR}/launcher/silu_and_mul.cu" + "${CMAKE_CURRENT_LIST_DIR}/launcher/readout_logprobs.cu" "${CMAKE_CURRENT_LIST_DIR}/launcher/target_logprobs.cu" "${CMAKE_CURRENT_LIST_DIR}/launcher/vision_pos_embed.cu" "${CMAKE_CURRENT_LIST_DIR}/wrapper/add_bias.cpp" @@ -70,6 +71,7 @@ target_sources(ninfer_ops PRIVATE "${CMAKE_CURRENT_LIST_DIR}/moe_experts/moe_expert_cpu.cpp" "${CMAKE_CURRENT_LIST_DIR}/moe_experts/gguf_expert_cpu.cpp" "${CMAKE_CURRENT_LIST_DIR}/wrapper/silu_mul.cpp" + "${CMAKE_CURRENT_LIST_DIR}/wrapper/readout_logprobs.cpp" "${CMAKE_CURRENT_LIST_DIR}/wrapper/target_logprobs.cpp" "${CMAKE_CURRENT_LIST_DIR}/wrapper/vision_pos_embed.cpp" ) diff --git a/src/ops/kernel/readout_logprobs.cuh b/src/ops/kernel/readout_logprobs.cuh new file mode 100644 index 0000000000..dfd90497a4 --- /dev/null +++ b/src/ops/kernel/readout_logprobs.cuh @@ -0,0 +1,52 @@ +#pragma once + +// Implements: include/ninfer/ops/readout_logprobs.h +// Match: contiguous BF16 [physical_rows], I32 [capacity], and FP32 [capacity]. +// Algorithm assumptions: one 1024-thread CTA performs a single-pass online logsumexp over the +// column, then thread k writes token k's log probability. + +#include "ops/common/warp.cuh" +#include "ops/kernel/target_logprobs.cuh" + +#include +#include + +#include + +namespace ninfer::ops { + +inline constexpr int kReadoutLogprobsBlock = 1024; + +template +__launch_bounds__(BlockSize) __global__ + void readout_logprobs_kernel(const __nv_bfloat16* logits, const std::int32_t* token_ids, + float* output, std::int32_t count, std::int32_t valid_rows) { + float local_max = -CUDART_INF_F; + float local_sum = 0.0f; + // Requires finite logits (see readout_logprobs.h): -inf before any finite value in a + // thread's subsequence NaN-poisons local_sum through expf(-inf - -inf). + for (std::int32_t row = static_cast(threadIdx.x); row < valid_rows; + row += BlockSize) { + const float value = __bfloat162float(logits[row]); + const float new_max = fmaxf(local_max, value); + if (new_max > local_max) { + local_sum *= expf(local_max - new_max); + } + local_max = new_max; + local_sum += expf(value - local_max); + } + const float maximum = target_logprobs_block_max(local_max); + + local_sum *= expf(local_max - maximum); + __shared__ float warp_sums[BlockSize / kWarpSize]; + // block_reduce_sum leaves the total in lane 0 of warp 0; every reader (k < count <= 32) + // sits in warp 0, so one shuffle broadcasts it there. + const float sum = block_reduce_sum(local_sum, warp_sums); + const float lse = maximum + logf(__shfl_sync(kFullWarpMask, sum, 0)); + const auto k = static_cast(threadIdx.x); + if (k < count) { + output[k] = __bfloat162float(logits[token_ids[k]]) - lse; + } +} + +} // namespace ninfer::ops diff --git a/src/ops/launcher/readout_logprobs.cu b/src/ops/launcher/readout_logprobs.cu new file mode 100644 index 0000000000..e9ab8b1af7 --- /dev/null +++ b/src/ops/launcher/readout_logprobs.cu @@ -0,0 +1,20 @@ +// Implements: include/ninfer/ops/readout_logprobs.h +// Match: wrapper-validated contiguous tensors and valid vocabulary rows. +// Algorithm assumptions: one CTA reads the whole column; no global workspace. +#include "ops/launcher/readout_logprobs.h" + +#include "core/device.h" +#include "ops/kernel/readout_logprobs.cuh" + +namespace ninfer::ops::detail { + +void readout_logprobs_launch(const Tensor& logits, const Tensor& token_ids, std::int32_t count, + std::int32_t valid_rows, Tensor& output, cudaStream_t stream) { + readout_logprobs_kernel<<<1, kReadoutLogprobsBlock, 0, stream>>>( + static_cast(logits.data), + static_cast(token_ids.data), static_cast(output.data), count, + valid_rows); + CUDA_CHECK(cudaGetLastError()); +} + +} // namespace ninfer::ops::detail diff --git a/src/ops/launcher/readout_logprobs.h b/src/ops/launcher/readout_logprobs.h new file mode 100644 index 0000000000..24ce3502c3 --- /dev/null +++ b/src/ops/launcher/readout_logprobs.h @@ -0,0 +1,16 @@ +#pragma once + +// ninfer::ops::detail - private launch prototype for readout_logprobs. + +#include "core/tensor.h" + +#include + +#include + +namespace ninfer::ops::detail { + +void readout_logprobs_launch(const Tensor& logits, const Tensor& token_ids, std::int32_t count, + std::int32_t valid_rows, Tensor& output, cudaStream_t stream); + +} // namespace ninfer::ops::detail diff --git a/src/ops/wrapper/readout_logprobs.cpp b/src/ops/wrapper/readout_logprobs.cpp new file mode 100644 index 0000000000..90e818d599 --- /dev/null +++ b/src/ops/wrapper/readout_logprobs.cpp @@ -0,0 +1,82 @@ +// ninfer::ops - readout_logprobs wrapper: public contract validation and launcher dispatch. +#include "ninfer/ops/readout_logprobs.h" + +#include "ops/launcher/readout_logprobs.h" + +#include +#include +#include +#include + +namespace ninfer::ops { +namespace { + +void require_vector(const Tensor& tensor, const char* label) { + if (tensor.ne[0] <= 0 || tensor.ne[1] != 1 || tensor.ne[2] != 1 || tensor.ne[3] != 1) { + throw std::invalid_argument(std::string("readout_logprobs: ") + label + + " must be a vector with a positive length"); + } +} + +void require_accessible(const Tensor& tensor, std::size_t alignment, const char* label) { + if (!tensor.is_contiguous()) { + throw std::invalid_argument(std::string("readout_logprobs: ") + label + + " must be contiguous"); + } + if (tensor.data == nullptr) { + throw std::invalid_argument(std::string("readout_logprobs: ") + label + + " data must be non-null"); + } + if ((reinterpret_cast(tensor.data) & (alignment - 1)) != 0) { + throw std::invalid_argument(std::string("readout_logprobs: ") + label + + " data is not naturally aligned"); + } +} + +bool overlaps(const Tensor& lhs, const Tensor& rhs) { + const auto lhs_begin = reinterpret_cast(lhs.data); + const auto rhs_begin = reinterpret_cast(rhs.data); + if (lhs_begin <= rhs_begin) { return rhs_begin - lhs_begin < lhs.bytes(); } + return lhs_begin - rhs_begin < rhs.bytes(); +} + +} // namespace + +void readout_logprobs(const Tensor& logits, const Tensor& token_ids, std::int32_t count, + std::int32_t valid_rows, Tensor& output, cudaStream_t stream) { + if (logits.dtype != DType::BF16) { + throw std::invalid_argument("readout_logprobs: logits must be BF16"); + } + if (token_ids.dtype != DType::I32) { + throw std::invalid_argument("readout_logprobs: token_ids must be I32"); + } + if (output.dtype != DType::FP32) { + throw std::invalid_argument("readout_logprobs: output must be FP32"); + } + + require_vector(logits, "logits"); + require_vector(token_ids, "token_ids"); + require_vector(output, "output"); + if (valid_rows <= 0 || valid_rows > logits.ne[0]) { + throw std::invalid_argument("readout_logprobs: valid_rows must be in [1, physical_rows]"); + } + if (count <= 0 || count > token_ids.ne[0] || count > output.ne[0] || + count > kReadoutLogprobsMaximum) { + throw std::invalid_argument( + "readout_logprobs: count must be in [1, min(capacity, kReadoutLogprobsMaximum)]"); + } + + (void)logits.bytes(); + (void)token_ids.bytes(); + (void)output.bytes(); + require_accessible(logits, alignof(std::uint16_t), "logits"); + require_accessible(token_ids, alignof(std::int32_t), "token_ids"); + require_accessible(output, alignof(float), "output"); + if (overlaps(output, logits) || overlaps(output, token_ids)) { + throw std::invalid_argument("readout_logprobs: output must not overlap either input"); + } + + detail::readout_logprobs_launch(logits, token_ids, count, valid_rows, output, stream); +} + +} // namespace ninfer::ops diff --git a/src/runtime/contract/execution.h b/src/runtime/contract/execution.h index 2b06a396f2..dd2740fe9a 100644 --- a/src/runtime/contract/execution.h +++ b/src/runtime/contract/execution.h @@ -62,6 +62,10 @@ struct RawTokenLogprob { float logprob = 0.0f; std::array top_ids{}; std::array top_values{}; + // The typed readout (ExecutionOptions::readout_tokens) on the round's first token: the first + // readout_count values are the named tokens' log probabilities in the request's order. + std::uint32_t readout_count = 0; + std::array readout_values{}; }; struct GeneratedRound { diff --git a/src/runtime/contract/request.h b/src/runtime/contract/request.h index e47e8d08b6..44203c908e 100644 --- a/src/runtime/contract/request.h +++ b/src/runtime/contract/request.h @@ -26,6 +26,9 @@ struct ResolvedExecutionOptions { ThinkingControlOptions thinking; // Gather each generated token's log probability and its likely alternatives. bool logprobs = false; + // With logprobs: the token ids whose exact log probabilities the first generated token's + // record carries (ExecutionOptions::readout_tokens). + std::vector readout_tokens; }; struct ResolvedRequestOptions { diff --git a/src/runtime/engine/engine.cpp b/src/runtime/engine/engine.cpp index 94081fe9ca..8fdf3d9368 100644 --- a/src/runtime/engine/engine.cpp +++ b/src/runtime/engine/engine.cpp @@ -110,6 +110,13 @@ runtime::ResolvedRequestOptions resolve_request_options(const ModelSamplingDefau throw std::invalid_argument("thinking budget must be positive"); } text::validate_structured_output(options.execution.structured_output); + if (options.execution.readout_tokens.size() > kMaximumReadoutTokens) { + throw std::invalid_argument("readout_tokens names more than " + + std::to_string(kMaximumReadoutTokens) + " tokens"); + } + if (!options.execution.readout_tokens.empty() && !options.execution.logprobs) { + throw std::invalid_argument("readout_tokens requires logprobs"); + } runtime::ResolvedRequestOptions resolved; resolved.execution.sampling = runtime::resolve_sampling(defaults, mode, options.execution.sampling); @@ -124,6 +131,7 @@ runtime::ResolvedRequestOptions resolve_request_options(const ModelSamplingDefau resolved.execution.thinking = options.execution.thinking; resolved.execution.structured_output = std::move(options.execution.structured_output); resolved.execution.logprobs = options.execution.logprobs; + resolved.execution.readout_tokens = std::move(options.execution.readout_tokens); resolved.stop = std::move(options.stop); resolved.output = options.output; resolved.ngram_session = std::move(options.ngram_session); diff --git a/src/serve/CMakeLists.txt b/src/serve/CMakeLists.txt index 609c2aac1e..be11845626 100644 --- a/src/serve/CMakeLists.txt +++ b/src/serve/CMakeLists.txt @@ -8,6 +8,8 @@ add_library(ninfer_serve STATIC openai_chat_http.cpp text_completion.cpp rerank.cpp + systemone.cpp + decisions_http.cpp llamacpp_http.cpp openai_common.cpp openai_responses_request.cpp diff --git a/src/serve/decisions_http.cpp b/src/serve/decisions_http.cpp new file mode 100644 index 0000000000..aa1000917b --- /dev/null +++ b/src/serve/decisions_http.cpp @@ -0,0 +1,112 @@ +// HTTP handling of the typed decision endpoints: POST /v1/systemone and POST /v1/decisions, one +// read per question on the served chat model. +#include "serve/http_server.h" +#include "serve/http_transport.h" +#include "serve/request_events.h" +#include "serve/request_validation.h" +#include "serve/systemone.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ninfer::serve { + +void HttpServer::handle_decisions(const httplib::Request& req, httplib::Response& res, + DecisionDialect dialect) { + const bool system_one = dialect == DecisionDialect::SystemOne; + const char* protocol = system_one ? "systemone" : "openai_decisions"; + DecisionRequest request; + ModelRegistry::Lease lease; + // The questions' labels as the model's tokens: each must be one token at the reply's start. + std::vector labels; + std::vector> label_tokens; + try { + const RequestJson body = parse_json_body(req); + request = system_one ? parse_systemone_request(body) : parse_decisions_request(body); + lease = lease_model(request.model, req); + request.model = lease.model(); + std::map resolved; + for (const DecisionQuestion& question : request.questions) { + labels.push_back(decision_labels(question)); + std::vector tokens; + for (const std::string& label : labels.back().texts) { + auto found = resolved.find(label); + if (found == resolved.end()) { + const std::vector ids = lease.service().tokenize(label, false); + if (ids.size() != 1) { + bad_request("label " + label + " is not one token of the model's tokenizer", + "questions"); + } + found = resolved.emplace(label, ids.front()).first; + } + tokens.push_back(found->second); + } + label_tokens.push_back(std::move(tokens)); + } + } catch (const ApiException& exception) { + write_openai_error(res, exception.error()); + return; + } + GenerationService& service = lease.service(); + const auto cancelled = [&req] { return client_disconnected(req); }; + // One read per question, submitted a lane's worth at a time: the Engine runs a window + // together, the reads of one window share the state's cached prefix, and the ingress queue + // keeps room for other clients' requests. + const std::size_t window = + std::max(1, static_cast(service.options().max_concurrency)); + std::vector answers; + answers.reserve(request.questions.size()); + int input_tokens = 0; + int output_tokens = 0; + try { + for (std::size_t begin = 0; begin < request.questions.size(); begin += window) { + const std::size_t end = std::min(begin + window, request.questions.size()); + std::vector reads; + std::vector batch; + std::vector> lifecycles; + for (std::size_t index = begin; index < end; ++index) { + reads.push_back(make_decision_read(request, request.questions[index], + label_tokens[index])); + batch.push_back(service.prepare(reads.back(), GenerationConsumerMode::Aggregate, + {.phase_timings = true}, cancelled)); + const RequestLogMetadata metadata{.model = request.model}; + lifecycles.push_back(begin_request(make_request_log_context( + ++request_seq_, protocol, reads.back(), metadata, batch.back()))); + } + for (std::size_t offset = 0; offset < batch.size(); ++offset) { + GenerationOutcome outcome; + try { + outcome = service.run(batch[offset], nullptr, cancelled); + } catch (const ApiException& exception) { + lifecycles[offset]->failure(make_generation_request_failure(exception.error())); + throw; + } + lifecycles[offset]->done(outcome); + input_tokens += outcome.prompt_tokens; + output_tokens += outcome.completion_tokens; + answers.push_back( + read_decision(outcome, labels[begin + offset], label_tokens[begin + offset])); + } + } + } catch (const ApiException& exception) { + write_openai_error(res, exception.error()); + return; + } catch (const std::exception& exception) { + ApiError error; + error.status = 500; + error.type = "internal_error"; + error.message = exception.what(); + write_openai_error(res, error); + return; + } + res.set_content(make_decision_response(request, answers, input_tokens, output_tokens), + "application/json"); +} + +} // namespace ninfer::serve diff --git a/src/serve/http_server.cpp b/src/serve/http_server.cpp index a506191ed9..2b7618ca5d 100644 --- a/src/serve/http_server.cpp +++ b/src/serve/http_server.cpp @@ -187,6 +187,8 @@ const char* endpoint_name(std::string_view request_path) noexcept { path == "/v1/reranking") { return "rerank"; } + if (path == "/v1/systemone") { return "systemone"; } + if (path == "/v1/decisions") { return "openai_decisions"; } if (path == "/v1/responses") { return "openai_responses"; } if (path == "/v1/responses/input_tokens") { return "openai_responses_input_tokens"; } if (path == "/v1/messages") { return "anthropic_messages"; } @@ -695,6 +697,14 @@ void HttpServer::register_routes() { register_post(endpoint, rerank); register_post(api_route_pattern(endpoint), rerank); } + register_post(api_route_pattern("/systemone"), + [this](const httplib::Request& req, httplib::Response& res) { + handle_decisions(req, res, DecisionDialect::SystemOne); + }); + register_post(api_route_pattern("/decisions"), + [this](const httplib::Request& req, httplib::Response& res) { + handle_decisions(req, res, DecisionDialect::OpenAI); + }); register_post(api_route_pattern("/responses"), [this](const httplib::Request& req, httplib::Response& res) { handle_responses(req, res); diff --git a/src/serve/http_server.h b/src/serve/http_server.h index f5292fbbad..f5be1e5ead 100644 --- a/src/serve/http_server.h +++ b/src/serve/http_server.h @@ -10,6 +10,7 @@ #include "serve/request_log.h" #include "serve/serve_metrics.h" #include "serve/serve_options.h" +#include "serve/systemone.h" #include "serve/text_completion.h" #include @@ -181,6 +182,9 @@ class HttpServer { TextCompletionDialect dialect); // POST /rerank, /reranking, /v1/rerank and /v1/reranking: the documents scored by the model. void handle_rerank(const httplib::Request& req, httplib::Response& res); + // POST /v1/systemone and /v1/decisions: typed questions about a state, one read each. + void handle_decisions(const httplib::Request& req, httplib::Response& res, + DecisionDialect dialect); void handle_props(const httplib::Request& req, httplib::Response& res) const; void handle_webui(const httplib::Request& req, httplib::Response& res) const; diff --git a/src/serve/openai_common.cpp b/src/serve/openai_common.cpp index 2ef01fb63e..a53bc64024 100644 --- a/src/serve/openai_common.cpp +++ b/src/serve/openai_common.cpp @@ -364,6 +364,8 @@ Json make_api_index(const std::string& model_id) { endpoint("POST", "/detokenize", "llama.cpp token ids back to text"), endpoint("POST", "/apply-template", "llama.cpp chat template rendering"), endpoint("POST", "/v1/rerank", "documents ranked by relevance to a query"), + endpoint("POST", "/v1/systemone", "typed decisions about a state (System One)"), + endpoint("POST", "/v1/decisions", "typed decisions about an input (OpenAI)"), endpoint("GET", "/models", "llama.cpp router: every configured model and its state"), endpoint("POST", "/models/load|unload|sleep", "llama.cpp router model lifecycle"), endpoint("GET", "/models/sse", "llama.cpp router model state events"), @@ -413,6 +415,8 @@ std::string new_openai_chat_completion_id() { return chat_identifier("chatcmpl-" std::string new_openai_completion_id() { return chat_identifier("cmpl-"); } +std::string new_openai_decision_id() { return chat_identifier("decision-"); } + std::string new_openai_chat_tool_call_id() { return chat_identifier("call_"); } std::string new_openai_request_id() { return responses_identifier("req"); } diff --git a/src/serve/openai_common.h b/src/serve/openai_common.h index ab01274c8a..e7bf3a2df3 100644 --- a/src/serve/openai_common.h +++ b/src/serve/openai_common.h @@ -76,6 +76,7 @@ void validate_openai_model(std::string_view requested, std::string_view availabl std::string new_openai_chat_completion_id(); std::string new_openai_completion_id(); +std::string new_openai_decision_id(); std::string new_openai_chat_tool_call_id(); std::string new_openai_request_id(); std::string new_openai_response_id(); diff --git a/src/serve/request.h b/src/serve/request.h index 59ce98c97e..d05ed8f494 100644 --- a/src/serve/request.h +++ b/src/serve/request.h @@ -286,6 +286,9 @@ struct GenerationRequest { // kMaximumTokenLogprobs) of its most likely alternatives in the response. bool logprobs = false; int top_logprobs = 0; + // With logprobs: token ids whose exact log probabilities the first content token's record + // carries (ninfer::ExecutionOptions::readout_tokens), the typed decisions' label readout. + std::vector readout_tokens; // The alternatives each reported token carries, absent unless the request asked for logprobs. [[nodiscard]] std::optional reported_top_logprobs() const noexcept { diff --git a/src/serve/systemone.cpp b/src/serve/systemone.cpp new file mode 100644 index 0000000000..5c676e443e --- /dev/null +++ b/src/serve/systemone.cpp @@ -0,0 +1,670 @@ +#include "serve/systemone.h" + +#include "serve/generation_service.h" +#include "serve/openai_common.h" +#include "serve/request_validation.h" + +#include + +#include +#include +#include +#include +#include +#include + +namespace ninfer::serve { +namespace { + +using Json = RequestJson; +using OutJson = nlohmann::ordered_json; + +// The system turn of every read. It keeps the reply's first token on the labels: on a 27B chat +// model the labels' probability mass rose from 0.6-0.9 to 0.98-0.99 with it. +constexpr const char* kSystem = + "You answer typed questions about a state with one label only: the letter of the best " + "option, yes or no, or the number of the level. No words before or after the label."; + +bool present(const Json& body, const char* key) { + return body.contains(key) && !body.at(key).is_null(); +} + +[[noreturn]] void not_implemented(std::string message, std::string param) { + ApiError error; + error.status = 501; + error.type = "not_implemented_error"; + error.message = std::move(message); + error.param = std::move(param); + throw ApiException(std::move(error)); +} + +// A string as given, other JSON as its compact text. +std::string json_text(const Json& value) { + if (value.is_string()) { return value.get(); } + return value.dump(-1, ' ', false, Json::error_handler_t::replace); +} + +std::string required_text(const Json& object, const char* key, const std::string& param) { + if (!present(object, key)) { bad_request(param + " is required", param); } + const Json& value = object.at(key); + if (!value.is_string() && !value.is_object() && !value.is_array()) { + bad_request(param + " must be a string, an object or an array", param); + } + return json_text(value); +} + +std::optional optional_description(const Json& value, const std::string& param) { + if (value.is_null()) { return std::nullopt; } + if (!value.is_string()) { bad_request(param + " must be a string or null", param); } + return value.get(); +} + +void refuse_unknown_fields(const Json& object, std::initializer_list known, + const std::string& scope) { + for (const auto& [key, value] : object.items()) { + if (std::find_if(known.begin(), known.end(), + [&key](const char* name) { return key == name; }) == known.end()) { + bad_request("unknown field " + scope + key, scope + key); + } + } +} + +void refuse_media(const Json& body) { + for (const char* key : {"images", "files", "audio", "videos"}) { + if (present(body, key)) { + not_implemented(std::string(key) + " are not supported: decisions read text", key); + } + } +} + +DecisionType parse_type(const Json& question, const std::string& param, DecisionDialect dialect) { + const std::string field = param + ".type"; + if (!present(question, "type") || !question.at("type").is_string()) { + bad_request(field + " must be a string", field); + } + const std::string type = question.at("type").get(); + if (type == "choice") { return DecisionType::Choice; } + if (type == "score") { return DecisionType::Score; } + if (dialect == DecisionDialect::SystemOne && type == "noul") { return DecisionType::Noul; } + if (dialect == DecisionDialect::OpenAI && type == "predicate") { return DecisionType::Noul; } + bad_request(field + " must be " + + (dialect == DecisionDialect::SystemOne ? "choice, noul or score" + : "predicate, choice or score"), + field); +} + +void require_option_count(std::size_t count, DecisionType type, const std::string& field) { + if (type == DecisionType::Choice && + (count < 2 || count > kMaximumDecisionOptions)) { + bad_request(field + " must list 2 to " + std::to_string(kMaximumDecisionOptions) + + " options", + field); + } + if (type == DecisionType::Score && (count < 2 || count > kMaximumDecisionLevels)) { + bad_request(field + " must list 2 to " + std::to_string(kMaximumDecisionLevels) + + " levels", + field); + } +} + +// System One's criteria: a choice's {option: description|null}, a score's [level, ...], a noul's +// optional {true: ..., false: ...}. +std::vector parse_criteria(const Json& question, DecisionType type, + const std::string& param) { + const std::string field = param + ".criteria"; + std::vector options; + switch (type) { + case DecisionType::Choice: { + if (!present(question, "criteria") || !question.at("criteria").is_object()) { + bad_request(field + " must map each option to its description or null", field); + } + const Json& criteria = question.at("criteria"); + require_option_count(criteria.size(), type, field); + for (const auto& [name, description] : criteria.items()) { + if (name.empty()) { bad_request(field + " has an empty option name", field); } + options.push_back({.value = Json(name), + .text = name, + .description = optional_description(description, field + "." + name)}); + } + break; + } + case DecisionType::Score: { + if (!present(question, "criteria") || !question.at("criteria").is_array()) { + bad_request(field + " must be an array of level descriptions, lowest first", field); + } + const Json& criteria = question.at("criteria"); + require_option_count(criteria.size(), type, field); + for (const Json& level : criteria) { + if (!level.is_string() || level.get_ref().empty()) { + bad_request(field + " levels must be non-empty strings", field); + } + options.push_back({.value = Json(options.size()), + .text = level.get(), + .description = std::nullopt}); + } + break; + } + case DecisionType::Noul: { + DecisionOption yes{.value = Json("yes"), .text = "yes", .description = std::nullopt}; + DecisionOption no{.value = Json("no"), .text = "no", .description = std::nullopt}; + if (present(question, "criteria")) { + const Json& criteria = question.at("criteria"); + if (!criteria.is_object()) { + bad_request(field + " must be null or an object with true and false descriptions", + field); + } + refuse_unknown_fields(criteria, {"true", "false"}, field + "."); + if (criteria.contains("true")) { + yes.description = optional_description(criteria.at("true"), field + ".true"); + } + if (criteria.contains("false")) { + no.description = optional_description(criteria.at("false"), field + ".false"); + } + } + options.push_back(std::move(yes)); + options.push_back(std::move(no)); + break; + } + } + return options; +} + +// OpenAI's choices [{value, description}] and levels [{label, description}]. +std::vector parse_openai_options(const Json& question, DecisionType type, + const std::string& param) { + std::vector options; + const bool choice = type == DecisionType::Choice; + const std::string field = param + (choice ? ".choices" : ".levels"); + if (type == DecisionType::Noul) { + if (present(question, "choices") || present(question, "levels")) { + bad_request(param + " is a predicate and takes neither choices nor levels", param); + } + options.push_back({.value = Json("yes"), .text = "yes", .description = std::nullopt}); + options.push_back({.value = Json("no"), .text = "no", .description = std::nullopt}); + return options; + } + if (present(question, choice ? "levels" : "choices")) { + bad_request(param + " takes " + (choice ? "choices, not levels" : "levels, not choices"), + param); + } + if (!present(question, choice ? "choices" : "levels") || + !question.at(choice ? "choices" : "levels").is_array()) { + bad_request(field + " must be an array", field); + } + const Json& entries = question.at(choice ? "choices" : "levels"); + require_option_count(entries.size(), type, field); + for (const Json& entry : entries) { + const std::string scope = field + "[" + std::to_string(options.size()) + "]"; + if (!entry.is_object()) { bad_request(scope + " must be an object", scope); } + refuse_unknown_fields(entry, {choice ? "value" : "label", "description"}, scope + "."); + DecisionOption option; + if (choice) { + if (!present(entry, "value")) { bad_request(scope + ".value is required", scope); } + const Json& value = entry.at("value"); + if (value.is_string()) { + option.text = value.get(); + } else if (value.is_boolean()) { + option.text = value.get() ? "true" : "false"; + } else { + bad_request(scope + ".value must be a string or a boolean", scope + ".value"); + } + option.value = value; + } else { + if (!present(entry, "label") || !entry.at("label").is_string()) { + bad_request(scope + ".label must be a string", scope + ".label"); + } + option.text = entry.at("label").get(); + option.value = Json(options.size()); + } + if (option.text.empty()) { bad_request(scope + " has an empty value", scope); } + if (entry.contains("description")) { + option.description = + optional_description(entry.at("description"), scope + ".description"); + } + for (const DecisionOption& earlier : options) { + if (earlier.text == option.text) { + bad_request(scope + " repeats " + option.text, scope); + } + } + options.push_back(std::move(option)); + } + return options; +} + +void require_question_count(std::size_t count, const char* param) { + if (count == 0) { bad_request(std::string(param) + " must hold at least one question", param); } + if (count > kMaximumDecisionQuestions) { + bad_request(std::string(param) + " holds more than " + + std::to_string(kMaximumDecisionQuestions) + " questions", + param); + } +} + +std::string openai_input_text(const Json& input) { + if (input.is_string()) { return input.get(); } + if (!input.is_array() || input.empty()) { + bad_request("input must be a string or a non-empty array of user messages", "input"); + } + std::string text; + for (std::size_t index = 0; index < input.size(); ++index) { + const std::string scope = "input[" + std::to_string(index) + "]"; + const Json& message = input[index]; + if (!message.is_object()) { bad_request(scope + " must be a message object", scope); } + refuse_unknown_fields(message, {"role", "content", "type"}, scope + "."); + if (present(message, "type") && message.at("type") != "message") { + bad_request(scope + ".type must be message", scope + ".type"); + } + if (!present(message, "role") || message.at("role") != "user") { + bad_request(scope + ".role must be user", scope + ".role"); + } + if (!present(message, "content")) { + bad_request(scope + ".content is required", scope + ".content"); + } + const Json& content = message.at("content"); + std::string piece; + if (content.is_string()) { + piece = content.get(); + } else if (content.is_array()) { + for (std::size_t part = 0; part < content.size(); ++part) { + const std::string part_scope = scope + ".content[" + std::to_string(part) + "]"; + const Json& entry = content[part]; + if (!entry.is_object() || !present(entry, "type") || + entry.at("type") != "input_text") { + bad_request(part_scope + " must be an input_text part", part_scope); + } + refuse_unknown_fields(entry, {"type", "text"}, part_scope + "."); + if (!present(entry, "text") || !entry.at("text").is_string()) { + bad_request(part_scope + ".text must be a string", part_scope + ".text"); + } + if (!piece.empty()) { piece += '\n'; } + piece += entry.at("text").get(); + } + } else { + bad_request(scope + ".content must be a string or an array of input_text parts", + scope + ".content"); + } + if (!text.empty()) { text += "\n\n"; } + text += piece; + } + return text; +} + +ChatTurn text_turn(ChatRole role, std::string text) { + ChatTurn turn; + turn.role = role; + turn.content.push_back(ContentPart{.kind = ContentKind::Text, .text = std::move(text)}); + return turn; +} + +std::string label_line(const std::string& label, const DecisionOption& option) { + std::string line = label + ". " + option.text; + if (option.description && !option.description->empty()) { + line += " - " + *option.description; + } + return line + "\n"; +} + +// "A, B or C" for a few labels, "A to Z" for more. +std::string label_span(const std::vector& labels) { + if (labels.size() > 4) { return labels.front() + " to " + labels.back(); } + std::string span; + for (std::size_t k = 0; k < labels.size(); ++k) { + if (k > 0) { span += k + 1 == labels.size() ? " or " : ", "; } + span += labels[k]; + } + return span; +} + +// The label the prompt shows for each option: the first of its labels. +std::vector shown_labels(const DecisionQuestion& question) { + const DecisionLabels labels = decision_labels(question); + std::vector shown(question.options.size()); + for (std::size_t k = labels.texts.size(); k-- > 0;) { + shown[labels.option[k]] = labels.texts[k]; + } + return shown; +} + +double finite_probability(double value) { + return std::isfinite(value) && value > 0.0 ? value : 0.0; +} + +} // namespace + +DecisionRequest parse_systemone_request(const Json& body) { + if (!body.is_object()) { bad_request("request body must be a JSON object"); } + refuse_media(body); + refuse_unknown_fields( + body, {"model", "state", "questions", "instructions", "chat_template_kwargs", "joint"}, + ""); + if (present(body, "joint")) { + if (!body.at("joint").is_boolean()) { + bad_request("joint must be a boolean", "joint"); + } + if (body.at("joint").get()) { + bad_request("joint reads are not supported: questions are read one at a time", + "joint"); + } + } + DecisionRequest out; + out.dialect = DecisionDialect::SystemOne; + if (present(body, "model")) { + if (!body.at("model").is_string()) { bad_request("model must be a string", "model"); } + out.model = body.at("model").get(); + } + out.state = required_text(body, "state", "state"); + if (present(body, "instructions")) { out.instructions = json_text(body.at("instructions")); } + if (present(body, "chat_template_kwargs")) { + const Json& kwargs = body.at("chat_template_kwargs"); + if (!kwargs.is_object()) { + bad_request("chat_template_kwargs must be an object", "chat_template_kwargs"); + } + if (kwargs.contains("enable_thinking") && kwargs.at("enable_thinking") == true) { + bad_request("decisions are read with thinking off", + "chat_template_kwargs.enable_thinking"); + } + } + if (!present(body, "questions") || !body.at("questions").is_object()) { + bad_request("questions must be an object of question ids", "questions"); + } + const Json& questions = body.at("questions"); + require_question_count(questions.size(), "questions"); + for (const auto& [id, question] : questions.items()) { + const std::string param = "questions." + id; + if (id.empty() || id.size() > kMaximumDecisionIdLength) { + bad_request("question ids are 1 to " + std::to_string(kMaximumDecisionIdLength) + + " characters", + "questions"); + } + if (!question.is_object()) { bad_request(param + " must be an object", param); } + refuse_unknown_fields(question, {"type", "instructions", "criteria"}, param + "."); + DecisionQuestion parsed; + parsed.id = id; + parsed.type = parse_type(question, param, DecisionDialect::SystemOne); + parsed.instructions = required_text(question, "instructions", param + ".instructions"); + parsed.options = parse_criteria(question, parsed.type, param); + out.questions.push_back(std::move(parsed)); + } + return out; +} + +DecisionRequest parse_decisions_request(const Json& body) { + if (!body.is_object()) { bad_request("request body must be a JSON object"); } + refuse_media(body); + refuse_unknown_fields(body, {"model", "input", "questions", "safety_identifier"}, ""); + DecisionRequest out; + out.dialect = DecisionDialect::OpenAI; + if (present(body, "model")) { + if (!body.at("model").is_string()) { bad_request("model must be a string", "model"); } + out.model = body.at("model").get(); + } + if (!present(body, "input")) { bad_request("input is required", "input"); } + out.state = openai_input_text(body.at("input")); + if (!present(body, "questions") || !body.at("questions").is_array()) { + bad_request("questions must be an array", "questions"); + } + const Json& questions = body.at("questions"); + require_question_count(questions.size(), "questions"); + for (std::size_t index = 0; index < questions.size(); ++index) { + const std::string param = "questions[" + std::to_string(index) + "]"; + const Json& question = questions[index]; + if (!question.is_object()) { bad_request(param + " must be an object", param); } + refuse_unknown_fields(question, {"type", "name", "instructions", "choices", "levels"}, + param + "."); + DecisionQuestion parsed; + parsed.named = present(question, "name"); + if (parsed.named) { + if (!question.at("name").is_string()) { + bad_request(param + ".name must be a string or null", param + ".name"); + } + parsed.id = question.at("name").get(); + } + parsed.type = parse_type(question, param, DecisionDialect::OpenAI); + if (!present(question, "instructions") || !question.at("instructions").is_string()) { + bad_request(param + ".instructions must be a string", param + ".instructions"); + } + parsed.instructions = question.at("instructions").get(); + parsed.options = parse_openai_options(question, parsed.type, param); + out.questions.push_back(std::move(parsed)); + } + return out; +} + +DecisionLabels decision_labels(const DecisionQuestion& question) { + DecisionLabels labels; + const auto add = [&labels](std::string text, std::size_t option) { + labels.texts.push_back(std::move(text)); + labels.option.push_back(option); + }; + switch (question.type) { + case DecisionType::Choice: + for (std::size_t k = 0; k < question.options.size(); ++k) { + add(std::string(1, static_cast('A' + k)), k); + } + break; + case DecisionType::Score: + for (std::size_t k = 0; k < question.options.size(); ++k) { add(std::to_string(k), k); } + break; + case DecisionType::Noul: + // Both cases: a model that answers "Yes" is answering yes. + add("yes", 0); + add("Yes", 0); + add("no", 1); + add("No", 1); + break; + } + return labels; +} + +GenerationRequest make_decision_read(const DecisionRequest& request, + const DecisionQuestion& question, + std::vector label_tokens) { + const std::vector labels = shown_labels(question); + std::string text; + if (!request.instructions.empty()) { text += request.instructions + "\n\n"; } + text += "State:\n" + request.state + "\n\n"; + text += "Question: " + question.instructions + "\n"; + switch (question.type) { + case DecisionType::Choice: + text += "Options:\n"; + for (std::size_t k = 0; k < question.options.size(); ++k) { + text += label_line(labels[k], question.options[k]); + } + text += "Reply with a single letter (" + label_span(labels) + ") and nothing else."; + break; + case DecisionType::Noul: + for (std::size_t k = 0; k < question.options.size(); ++k) { + const DecisionOption& option = question.options[k]; + if (option.description && !option.description->empty()) { + text += labels[k] + ": " + *option.description + "\n"; + } + } + text += "Answer with one word, yes or no, in lower case."; + break; + case DecisionType::Score: + text += "Levels, lowest first:\n"; + for (std::size_t k = 0; k < question.options.size(); ++k) { + text += label_line(labels[k], question.options[k]); + } + text += "Reply with a single number (" + label_span(labels) + ") and nothing else."; + break; + } + GenerationRequest out; + out.messages.push_back(text_turn(ChatRole::System, kSystem)); + out.messages.push_back(text_turn(ChatRole::User, std::move(text))); + out.enable_thinking = false; + out.max_tokens = 1; + out.sampling.temperature = 0.0; + out.sampling.presence_penalty = 0.0; + out.sampling.frequency_penalty = 0.0; + out.logprobs = true; + out.readout_tokens = std::move(label_tokens); + // The server's default graft would be a hidden prefix the question does not expect. + out.graft = std::string(); + return out; +} + +DecisionAnswer read_decision(const GenerationOutcome& outcome, const DecisionLabels& labels, + std::span label_tokens) { + if (outcome.content_logprobs.empty() || outcome.content_logprobs.front().top_ids[0] < 0) { + throw std::runtime_error("the decision read produced no answer token"); + } + if (label_tokens.size() != labels.option.size() || labels.option.empty()) { + throw std::logic_error("decision label tokens do not match the labels"); + } + const ninfer::TokenLogprob& first = outcome.content_logprobs.front(); + const std::size_t count = label_tokens.size(); + const std::size_t options = 1 + *std::max_element(labels.option.begin(), labels.option.end()); + DecisionAnswer answer; + answer.exact = first.readout.size() == count; + std::vector logprobs(count, -std::numeric_limits::infinity()); + if (answer.exact) { + for (std::size_t k = 0; k < count; ++k) { logprobs[k] = first.readout[k]; } + } else { + // The top alternatives only: a label outside them counts as likely as the least likely of + // them, its upper bound. + double least = -std::numeric_limits::infinity(); + for (std::size_t j = 0; j < ninfer::kMaximumTokenLogprobs && first.top_ids[j] >= 0; ++j) { + least = first.top_values[j]; + for (std::size_t k = 0; k < count; ++k) { + if (first.top_ids[j] == label_tokens[k]) { logprobs[k] = least; } + } + } + for (double& value : logprobs) { + if (!std::isfinite(value)) { value = least; } + } + } + answer.probabilities.assign(options, 0.0); + for (std::size_t k = 0; k < count; ++k) { + const double probability = finite_probability(std::exp(logprobs[k])); + answer.probabilities[labels.option[k]] += probability; + answer.label_mass += probability; + } + if (answer.label_mass > 0.0) { + for (double& probability : answer.probabilities) { probability /= answer.label_mass; } + } else { + for (double& probability : answer.probabilities) { + probability = 1.0 / static_cast(options); + } + } + answer.argmax_is_label = + std::find(label_tokens.begin(), label_tokens.end(), first.top_ids[0]) != label_tokens.end(); + return answer; +} + +std::string make_decision_response(const DecisionRequest& request, + std::span answers, int input_tokens, + int output_tokens) { + if (answers.size() != request.questions.size()) { + throw std::logic_error("decision answers do not match the questions"); + } + const bool system_one = request.dialect == DecisionDialect::SystemOne; + OutJson body; + body["id"] = new_openai_decision_id(); + body["object"] = system_one ? "structured_decision" : "decision"; + body["created"] = unix_time_now(); + body["model"] = request.model; + OutJson rendered = system_one ? OutJson::object() : OutJson::array(); + OutJson diagnostics = OutJson::object(); + for (std::size_t q = 0; q < answers.size(); ++q) { + const DecisionQuestion& question = request.questions[q]; + const DecisionAnswer& answer = answers[q]; + const auto best = static_cast( + std::max_element(answer.probabilities.begin(), answer.probabilities.end()) - + answer.probabilities.begin()); + const double confidence = answer.probabilities[best] * answer.label_mass; + OutJson entry; + if (system_one) { + switch (question.type) { + case DecisionType::Choice: { + OutJson probabilities = OutJson::object(); + for (std::size_t k = 0; k < question.options.size(); ++k) { + probabilities[question.options[k].text] = answer.probabilities[k]; + } + entry = {{"type", "choice"}, + {"choice", question.options[best].text}, + {"probabilities", std::move(probabilities)}, + {"confidence", confidence}}; + break; + } + case DecisionType::Noul: + entry = {{"type", "noul"}, + {"noul", answer.probabilities[0]}, + {"probabilities", + {{"yes", answer.probabilities[0]}, {"no", answer.probabilities[1]}}}, + {"confidence", confidence}}; + break; + case DecisionType::Score: { + OutJson legend = OutJson::object(); + OutJson probabilities = OutJson::object(); + double score = 0.0; + for (std::size_t k = 0; k < question.options.size(); ++k) { + const std::string index = std::to_string(k); + legend[index] = question.options[k].text; + probabilities[index] = answer.probabilities[k]; + score += answer.probabilities[k] * static_cast(k); + } + entry = {{"type", "score"}, + {"score", score}, + {"legend", std::move(legend)}, + {"probabilities", std::move(probabilities)}, + {"confidence", confidence}}; + break; + } + } + rendered[question.id] = std::move(entry); + diagnostics[question.id] = {{"label_mass", answer.label_mass}, + {"argmax_is_label", answer.argmax_is_label}, + {"readout", answer.exact ? "exact" : "top_k"}}; + continue; + } + const OutJson name = question.named ? OutJson(question.id) : OutJson(nullptr); + switch (question.type) { + case DecisionType::Noul: + entry = {{"type", "predicate"}, {"name", name}, {"probability", answer.probabilities[0]}}; + break; + case DecisionType::Choice: { + OutJson probabilities = OutJson::array(); + for (std::size_t k = 0; k < question.options.size(); ++k) { + probabilities.push_back({{"value", question.options[k].value}, + {"probability", answer.probabilities[k]}}); + } + entry = {{"type", "choice"}, + {"name", name}, + {"choice", question.options[best].value}, + {"probabilities", std::move(probabilities)}, + {"confidence", confidence}}; + break; + } + case DecisionType::Score: { + OutJson probabilities = OutJson::array(); + double score = 0.0; + for (std::size_t k = 0; k < question.options.size(); ++k) { + probabilities.push_back({{"value", k}, + {"label", question.options[k].text}, + {"probability", answer.probabilities[k]}}); + score += answer.probabilities[k] * static_cast(k); + } + entry = {{"type", "score"}, + {"name", name}, + {"score", score}, + {"probabilities", std::move(probabilities)}, + {"confidence", confidence}}; + break; + } + } + rendered.push_back(std::move(entry)); + } + body["answers"] = std::move(rendered); + if (system_one) { + body["usage"] = {{"input_tokens", input_tokens}, {"output_tokens", output_tokens}}; + body["diagnostics"] = std::move(diagnostics); + } else { + body["usage"] = {{"input_tokens", input_tokens}, + {"output_tokens", output_tokens}, + {"total_tokens", input_tokens + output_tokens}}; + } + return body.dump(-1, ' ', false, OutJson::error_handler_t::replace); +} + +} // namespace ninfer::serve diff --git a/src/serve/systemone.h b/src/serve/systemone.h new file mode 100644 index 0000000000..638bb0df60 --- /dev/null +++ b/src/serve/systemone.h @@ -0,0 +1,112 @@ +#pragma once + +// Typed decisions: POST /v1/systemone (TypeSafe's System One API, as llama.cpp, vLLM and Ollama +// serve it) and POST /v1/decisions (OpenAI's Decisions API), answered by the served chat model +// through a label readout. Each question is one prompt -- a fixed system turn asking for one +// label, then the state, the question and its options under one-token labels (A.. for a choice, +// yes/no for a noul, 0.. for a score), thinking off -- and the answer is read from the first reply +// token's exact log probabilities of those labels (a noul's yes and no in both cases): softmax +// over the options gives the probabilities, their total over the vocabulary is the label mass, +// and the winner's probability over the whole vocabulary is the confidence. + +#include "serve/request.h" +#include "serve/request_json.h" + +#include +#include +#include +#include +#include + +namespace ninfer::serve { + +struct GenerationOutcome; + +// vLLM's bounds, within the readout's kMaximumReadoutTokens and the 26 letters. +inline constexpr std::size_t kMaximumDecisionQuestions = 64; +inline constexpr std::size_t kMaximumDecisionOptions = 26; +inline constexpr std::size_t kMaximumDecisionLevels = 10; +inline constexpr std::size_t kMaximumDecisionIdLength = 100; + +enum class DecisionDialect { SystemOne, OpenAI }; + +// System One's choice / noul / score; OpenAI's predicate is a noul. +enum class DecisionType { Choice, Noul, Score }; + +struct DecisionOption { + // The option's wire value: a System One option name or score level, or an OpenAI choice value + // (a string or a boolean, echoed with its type) or level label. + RequestJson value; + // The value as the prompt shows it. + std::string text; + std::optional description; +}; + +struct DecisionQuestion { + // System One's question id, or OpenAI's optional name (`named` false echoes null). + std::string id; + bool named = true; + DecisionType type = DecisionType::Choice; + std::string instructions; + // A choice's options or a score's levels, lowest first; a noul's optional yes and no + // descriptions as two entries (text "yes" and "no"). + std::vector options; +}; + +struct DecisionRequest { + DecisionDialect dialect = DecisionDialect::SystemOne; + std::string model; + // The state as the prompt shows it: a string as given, other JSON as its compact text. + std::string state; + // Optional context placed ahead of the state (System One's `instructions`). + std::string instructions; + std::vector questions; +}; + +// Strict: unknown fields are refused, as vLLM refuses them; media fields are refused as not +// implemented. +[[nodiscard]] DecisionRequest parse_systemone_request(const RequestJson& body); +[[nodiscard]] DecisionRequest parse_decisions_request(const RequestJson& body); + +struct DecisionLabels { + // The label texts the read looks up, each one token: A..Z for a choice, yes, Yes, no and No + // for a noul, 0..9 for a score. + std::vector texts; + // The option each label answers for, aligned with texts. + std::vector option; +}; + +[[nodiscard]] DecisionLabels decision_labels(const DecisionQuestion& question); + +// The request that reads `question`: the system turn, then one user turn with the state, the +// question and its options under the labels, thinking off, one output token, with the labels' +// token ids (aligned with decision_labels(question).texts) read out exactly. +[[nodiscard]] GenerationRequest make_decision_read(const DecisionRequest& request, + const DecisionQuestion& question, + std::vector label_tokens); + +struct DecisionAnswer { + // Over the options, in order; sums to 1. + std::vector probabilities; + // The labels' total probability over the whole vocabulary. + double label_mass = 0.0; + // The model's most likely token is one of the labels. + bool argmax_is_label = false; + // Read from the exact readout; false when the core has none and the labels were read from the + // top alternatives (a label outside them counts as likely as the least likely alternative). + bool exact = true; +}; + +// The answer of one read, `label_tokens` aligned with `labels.texts`, or std::runtime_error when +// the read produced no answer token. +[[nodiscard]] DecisionAnswer read_decision(const GenerationOutcome& outcome, + const DecisionLabels& labels, + std::span label_tokens); + +// One response, the answers aligned with the request's questions; `input_tokens` and +// `output_tokens` summed over the reads. +[[nodiscard]] std::string make_decision_response(const DecisionRequest& request, + std::span answers, + int input_tokens, int output_tokens); + +} // namespace ninfer::serve diff --git a/src/serve/translate.cpp b/src/serve/translate.cpp index 239c51e346..0e5381e225 100644 --- a/src/serve/translate.cpp +++ b/src/serve/translate.cpp @@ -471,7 +471,8 @@ ninfer::RequestOptions to_request_options(const GenerationRequest& request, options.execution.post_thinking_sampling->presence_penalty = 0.0F; } } - options.execution.logprobs = request.logprobs; + options.execution.logprobs = request.logprobs; + options.execution.readout_tokens = request.readout_tokens; options.output.raw = false; options.output.preserve_special_tokens = request.structured_output.kind == StructuredOutputKind::None && diff --git a/tests/cmake/ProductTests.cmake b/tests/cmake/ProductTests.cmake index 3028342a3e..fa05e2d734 100644 --- a/tests/cmake/ProductTests.cmake +++ b/tests/cmake/ProductTests.cmake @@ -69,6 +69,10 @@ ninfer_add_test(ninfer_rerank_test SOURCES "${CMAKE_CURRENT_LIST_DIR}/../test_rerank.cpp" LIBRARIES ninfer_serve) +ninfer_add_test(ninfer_systemone_test + SOURCES "${CMAKE_CURRENT_LIST_DIR}/../test_systemone.cpp" + LIBRARIES ninfer_serve) + ninfer_add_test(ninfer_serve_metrics_test SOURCES "${CMAKE_CURRENT_LIST_DIR}/../test_serve_metrics.cpp" LIBRARIES ninfer_serve) diff --git a/tests/ops/test_readout_logprobs.cpp b/tests/ops/test_readout_logprobs.cpp new file mode 100644 index 0000000000..0599f0ce89 --- /dev/null +++ b/tests/ops/test_readout_logprobs.cpp @@ -0,0 +1,228 @@ +#include "ninfer/ops/readout_logprobs.h" +#include "ops/op_tester.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace ninfer; +using namespace ninfer::test; + +namespace { + +constexpr ReductionCriterion kReadoutLogprobsFp32Criterion{ + /*relative_l2=*/2.0e-5, + /*gross_absolute=*/2.0e-4, + /*gross_relative_to_max_reference=*/0.0, +}; + +// `count` ids: the first and last valid rows, then scattered rows. +std::vector make_ids(std::int32_t valid_rows, std::int32_t count) { + std::vector ids(static_cast(count)); + for (std::int32_t k = 0; k < count; ++k) { + if (k == 0) { + ids[static_cast(k)] = 0; + } else if (k == 1) { + ids[static_cast(k)] = valid_rows - 1; + } else { + ids[static_cast(k)] = static_cast( + (static_cast(k + 1) * 7919u) % static_cast(valid_rows)); + } + } + return ids; +} + +std::vector make_random_logits(std::int32_t physical_rows, std::int32_t valid_rows) { + std::vector logits(static_cast(physical_rows)); + for (std::int32_t row = 0; row < valid_rows; ++row) { + const std::uint32_t mixed = static_cast(row) * 1664525u + 1013904223u; + const float value = -24.0f + static_cast(mixed % 6144u) * (1.0f / 128.0f); + logits[static_cast(row)] = f32_to_bf16(value); + } + for (std::int32_t row = valid_rows; row < physical_rows; ++row) { + logits[static_cast(row)] = f32_to_bf16(96.0f); + } + return logits; +} + +std::vector make_uniform_logits(std::int32_t physical_rows, std::int32_t valid_rows, + float value) { + std::vector logits(static_cast(physical_rows), f32_to_bf16(value)); + for (std::int32_t row = valid_rows; row < physical_rows; ++row) { + logits[static_cast(row)] = f32_to_bf16(112.0f); + } + return logits; +} + +std::vector make_extreme_logits(std::int32_t physical_rows, std::int32_t valid_rows) { + std::vector logits(static_cast(physical_rows)); + constexpr float values[] = {-80.0f, -32.0f, -1.0f, 0.0f, 1.0f, 32.0f, 80.0f}; + for (std::int32_t row = 0; row < valid_rows; ++row) { + logits[static_cast(row)] = + f32_to_bf16(values[static_cast(row) % std::size(values)]); + } + for (std::int32_t row = valid_rows; row < physical_rows; ++row) { + logits[static_cast(row)] = f32_to_bf16(120.0f); + } + return logits; +} + +std::vector readout_oracle(const std::vector& logits, + const std::vector& ids, std::int32_t valid_rows) { + double maximum = -std::numeric_limits::infinity(); + for (std::int32_t row = 0; row < valid_rows; ++row) { + maximum = std::max(maximum, static_cast(bf16_to_f32(logits[row]))); + } + double sum = 0.0; + for (std::int32_t row = 0; row < valid_rows; ++row) { + sum += std::exp(static_cast(bf16_to_f32(logits[row])) - maximum); + } + std::vector expected(ids.size()); + for (std::size_t k = 0; k < ids.size(); ++k) { + expected[k] = static_cast(bf16_to_f32(logits[ids[k]])) - maximum - std::log(sum); + } + return expected; +} + +int run_case(const std::string& label, std::int32_t physical_rows, std::int32_t valid_rows, + std::int32_t count, std::int32_t capacity, const std::vector& logits) { + const auto ids = make_ids(valid_rows, count); + const auto expected = readout_oracle(logits, ids, valid_rows); + std::vector stored_ids(static_cast(capacity), -7); + std::copy(ids.begin(), ids.end(), stored_ids.begin()); + + GuardedDeviceBuffer device_logits(logits.size() * sizeof(std::uint16_t)); + GuardedDeviceBuffer device_ids(stored_ids.size() * sizeof(std::int32_t)); + GuardedDeviceBuffer device_output(static_cast(capacity) * sizeof(float)); + device_logits.copy_from_host(logits.data(), device_logits.bytes()); + device_ids.copy_from_host(stored_ids.data(), device_ids.bytes()); + device_output.fill(0xcd); + + Tensor logits_tensor(device_logits.data(), DType::BF16, {physical_rows}); + Tensor ids_tensor(device_ids.data(), DType::I32, {capacity}); + Tensor output_tensor(device_output.data(), DType::FP32, {capacity}); + ops::readout_logprobs(logits_tensor, ids_tensor, count, valid_rows, output_tensor, nullptr); + cuda_synchronize(); + + const auto written = from_device(device_output.data(), static_cast(count)); + int failures = verify_reduction(label, std::vector(written.begin(), written.end()), + expected, kReadoutLogprobsFp32Criterion); + // Slots past `count` keep their fill. + const auto raw = from_device(device_output.data(), static_cast(capacity)); + for (std::size_t k = static_cast(count); k < raw.size(); ++k) { + if (raw[k] != 0xcdcdcdcdu) { + std::cerr << label << ": output slot " << k << " past count was written\n"; + ++failures; + break; + } + } + failures += + verify_exact((label + " preserves logits").c_str(), + from_device(device_logits.data(), logits.size()), logits); + failures += verify_exact((label + " preserves ids").c_str(), + from_device(device_ids.data(), stored_ids.size()), + stored_ids); + failures += device_logits.verify_guards(label + " logits guards"); + failures += device_ids.verify_guards(label + " id guards"); + failures += device_output.verify_guards(label + " output guards"); + return failures; +} + +template +int expect_invalid(const char* label, Function&& function) { + try { + function(); + } catch (const std::invalid_argument&) { return 0; } catch (const std::exception& error) { + std::cerr << label << ": expected invalid_argument, got " << error.what() << '\n'; + return 1; + } + std::cerr << label << ": expected invalid_argument\n"; + return 1; +} + +int run_validation_cases() { + DeviceBuffer logits_data(16 * sizeof(std::uint16_t)); + DeviceBuffer ids_data(8 * sizeof(std::int32_t)); + DeviceBuffer output_data(8 * sizeof(float)); + Tensor logits(logits_data.p, DType::BF16, {16}); + Tensor ids(ids_data.p, DType::I32, {8}); + Tensor output(output_data.p, DType::FP32, {8}); + + int failures = 0; + failures += expect_invalid("readout_logprobs rejects count=0", + [&] { ops::readout_logprobs(logits, ids, 0, 16, output, nullptr); }); + failures += expect_invalid("readout_logprobs rejects count>capacity", + [&] { ops::readout_logprobs(logits, ids, 9, 16, output, nullptr); }); + failures += expect_invalid("readout_logprobs rejects count>maximum", [&] { + DeviceBuffer wide_ids(64 * sizeof(std::int32_t)); + DeviceBuffer wide_output(64 * sizeof(float)); + Tensor many_ids(wide_ids.p, DType::I32, {64}); + Tensor many_output(wide_output.p, DType::FP32, {64}); + ops::readout_logprobs(logits, many_ids, 33, 16, many_output, nullptr); + }); + failures += expect_invalid("readout_logprobs rejects valid_rows=0", + [&] { ops::readout_logprobs(logits, ids, 4, 0, output, nullptr); }); + failures += expect_invalid("readout_logprobs rejects valid_rows>physical_rows", + [&] { ops::readout_logprobs(logits, ids, 4, 17, output, nullptr); }); + failures += expect_invalid("readout_logprobs rejects output dtype", [&] { + Tensor wrong_output(output_data.p, DType::BF16, {8}); + ops::readout_logprobs(logits, ids, 4, 16, wrong_output, nullptr); + }); + failures += expect_invalid("readout_logprobs rejects id dtype", [&] { + Tensor wrong_ids(ids_data.p, DType::FP32, {8}); + ops::readout_logprobs(logits, wrong_ids, 4, 16, output, nullptr); + }); + failures += expect_invalid("readout_logprobs rejects a logits matrix", [&] { + Tensor matrix(logits_data.p, DType::BF16, {8, 2}); + ops::readout_logprobs(matrix, ids, 4, 8, output, nullptr); + }); + failures += expect_invalid("readout_logprobs rejects non-contiguous logits", [&] { + Tensor strided_logits = logits; + strided_logits.nb[0] += 2; + ops::readout_logprobs(strided_logits, ids, 4, 16, output, nullptr); + }); + failures += expect_invalid("readout_logprobs rejects null output", [&] { + Tensor null_output(nullptr, DType::FP32, {8}); + ops::readout_logprobs(logits, ids, 4, 16, null_output, nullptr); + }); + failures += expect_invalid("readout_logprobs rejects output alias", [&] { + Tensor alias_output(logits_data.p, DType::FP32, {8}); + ops::readout_logprobs(logits, ids, 4, 16, alias_output, nullptr); + }); + return failures; +} + +} // namespace + +int main() { + if (cuda_unavailable()) { + std::cout << "SKIP: no usable CUDA device\n"; + return 77; + } + + int failures = 0; + failures += run_case("readout_logprobs full vocabulary 32 ids", 248320, 248077, 32, 32, + make_random_logits(248320, 248077)); + failures += run_case("readout_logprobs Qwen vocabulary 26 ids", 151936, 151936, 26, 32, + make_random_logits(151936, 151936)); + failures += run_case("readout_logprobs non-aligned rows, one id", 523, 509, 1, 32, + make_random_logits(523, 509)); + failures += run_case("readout_logprobs fewer rows than threads", 263, 257, 10, 10, + make_random_logits(263, 257)); + failures += run_case("readout_logprobs uniform logits", 263, 257, 12, 32, + make_uniform_logits(263, 257, 3.5f)); + failures += run_case("readout_logprobs one valid row", 13, 1, 1, 4, + make_uniform_logits(13, 1, -7.0f)); + failures += run_case("readout_logprobs extreme finite logits", 4099, 4093, 32, 32, + make_extreme_logits(4099, 4093)); + failures += run_validation_cases(); + + std::cout << (failures ? "FAIL" : "OK") << " readout_logprobs\n"; + return failures ? 1 : 0; +} diff --git a/tests/ops/tests.cmake b/tests/ops/tests.cmake index d4ecac9a5a..5658ad7576 100644 --- a/tests/ops/tests.cmake +++ b/tests/ops/tests.cmake @@ -36,6 +36,7 @@ set(ninfer_op_tests scatter scatter_bf16_batch target_logprobs + readout_logprobs paged_kv_window logprob_topk position) diff --git a/tests/test_systemone.cpp b/tests/test_systemone.cpp new file mode 100644 index 0000000000..53d45558d7 --- /dev/null +++ b/tests/test_systemone.cpp @@ -0,0 +1,400 @@ +// Typed decisions on the wire, without an Engine: System One's and OpenAI's requests, the labels +// and the prompt each question becomes, the answer read from the first token's readout (or its +// top alternatives), and both responses. + +#include "serve/generation_service.h" +#include "serve/systemone.h" + +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +using namespace ninfer::serve; +using Json = nlohmann::json; + +int failures = 0; + +void expect(bool condition, const std::string& label) { + if (condition) { return; } + std::cerr << "expectation failed: " << label << '\n'; + ++failures; +} + +bool close_to(double a, double b) { return std::abs(a - b) < 1e-6; } + +struct Refusal { + int status = 0; + std::string param; +}; + +Refusal systemone_refusal(const char* body) { + try { + (void)parse_systemone_request(RequestJson::parse(body)); + } catch (const ApiException& error) { + return {.status = error.error().status, .param = error.error().param}; + } + return {}; +} + +Refusal decisions_refusal(const char* body) { + try { + (void)parse_decisions_request(RequestJson::parse(body)); + } catch (const ApiException& error) { + return {.status = error.error().status, .param = error.error().param}; + } + return {}; +} + +const char* kSystemOne = R"({ + "model": "m", + "state": {"ticket": "My card was charged twice for one order."}, + "questions": { + "team": {"type": "choice", "instructions": "Which team?", + "criteria": {"billing": "payments and refunds", "shipping": null}}, + "urgent": {"type": "noul", "instructions": "Reply within the hour?", + "criteria": {"true": "money is moving"}}, + "tone": {"type": "score", "instructions": "How upset?", + "criteria": ["calm", "annoyed", "furious"]} + } +})"; + +const char* kOpenAI = R"({ + "model": "m", + "input": [{"role": "user", "content": [{"type": "input_text", "text": "Broken screen."}]}, + {"role": "user", "content": "Second message."}], + "questions": [ + {"type": "predicate", "name": "damaged", "instructions": "Damaged item?"}, + {"type": "choice", "name": "department", "instructions": "Which department?", + "choices": [{"value": "returns", "description": "RMA"}, {"value": true}]}, + {"type": "score", "instructions": "How severe?", + "levels": [{"label": "cosmetic"}, {"label": "broken"}]} + ] +})"; + +// A first-token record whose readout holds the given label log probabilities and whose top +// alternatives are the given (id, probability) pairs, most likely first. +GenerationOutcome read(std::vector readout, + std::vector> top) { + ninfer::TokenLogprob first; + first.top_ids.fill(-1); + for (std::size_t k = 0; k < top.size(); ++k) { + first.top_ids[k] = top[k].first; + first.top_values[k] = static_cast(std::log(top[k].second)); + } + first.id = first.top_ids[0]; + first.logprob = first.top_values[0]; + first.readout.assign(readout.begin(), readout.end()); + GenerationOutcome out; + out.content_logprobs.push_back(std::move(first)); + out.prompt_tokens = 10; + out.completion_tokens = 1; + return out; +} + +void systemone_parsing() { + const DecisionRequest request = parse_systemone_request(RequestJson::parse(kSystemOne)); + expect(request.dialect == DecisionDialect::SystemOne && request.model == "m", + "System One dialect and model"); + expect(request.state == R"({"ticket":"My card was charged twice for one order."})", + "a JSON state becomes its compact text"); + expect(request.questions.size() == 3 && request.questions[0].id == "team" && + request.questions[1].id == "urgent" && request.questions[2].id == "tone", + "questions keep the request's order"); + const DecisionQuestion& team = request.questions[0]; + expect(team.type == DecisionType::Choice && team.options.size() == 2 && + team.options[0].text == "billing" && + team.options[0].description == "payments and refunds" && + team.options[1].text == "shipping" && !team.options[1].description, + "a choice's options with and without descriptions"); + const DecisionQuestion& urgent = request.questions[1]; + expect(urgent.type == DecisionType::Noul && urgent.options.size() == 2 && + urgent.options[0].text == "yes" && urgent.options[0].description == "money is moving" && + urgent.options[1].text == "no" && !urgent.options[1].description, + "a noul's yes and no with the true description"); + const DecisionQuestion& tone = request.questions[2]; + expect(tone.type == DecisionType::Score && tone.options.size() == 3 && + tone.options[2].text == "furious" && tone.options[2].value == Json(2), + "a score's levels, lowest first"); + const DecisionRequest plain = parse_systemone_request(RequestJson::parse( + R"({"state": "s", "instructions": "ctx", + "questions": {"q": {"type": "noul", "instructions": {"k": 1}}}})")); + expect(plain.state == "s" && plain.instructions == "ctx" && + plain.questions[0].instructions == R"({"k":1})" && + !plain.questions[0].options[0].description, + "a string state, context, JSON instructions, a noul without criteria"); + + expect(systemone_refusal(R"({"questions": {"q": {"type": "noul", "instructions": "i"}}})") + .param == "state", + "a state is required"); + expect(systemone_refusal(R"({"state": "s", "questions": {}})").param == "questions", + "at least one question"); + expect(systemone_refusal(R"({"state": "s", "extra": 1, "questions": {}})").param == "extra", + "unknown top-level fields are refused"); + expect(systemone_refusal( + R"({"state": "s", "questions": {"q": {"type": "noul", "instructions": "i", "x": 1}}})") + .param == "questions.q.x", + "unknown question fields are refused"); + expect(systemone_refusal( + R"({"state": "s", "questions": {"q": {"type": "choice", "instructions": "i", + "criteria": {"only": null}}}})") + .param == "questions.q.criteria", + "a choice needs two options"); + expect(systemone_refusal( + R"({"state": "s", "questions": {"q": {"type": "score", "instructions": "i", + "criteria": ["a"]}}})") + .param == "questions.q.criteria", + "a score needs two levels"); + expect(systemone_refusal( + R"({"state": "s", "questions": {"q": {"type": "predicate", "instructions": "i"}}})") + .param == "questions.q.type", + "OpenAI's predicate is not a System One type"); + expect(systemone_refusal( + R"({"state": "s", "joint": true, "questions": {"q": {"type": "noul", "instructions": "i"}}})") + .param == "joint", + "joint reads are refused"); + expect(systemone_refusal( + R"({"state": "s", "images": ["data:image/png;base64,AA=="], + "questions": {"q": {"type": "noul", "instructions": "i"}}})") + .status == 501, + "images are not implemented"); + std::string many = R"({"state": "s", "questions": {)"; + for (int q = 0; q < 65; ++q) { + many += (q ? "," : "") + std::string("\"q") + std::to_string(q) + + R"(": {"type": "noul", "instructions": "i"})"; + } + many += "}}"; + expect(systemone_refusal(many.c_str()).param == "questions", "at most 64 questions"); +} + +void decisions_parsing() { + const DecisionRequest request = parse_decisions_request(RequestJson::parse(kOpenAI)); + expect(request.dialect == DecisionDialect::OpenAI && request.model == "m", + "OpenAI dialect and model"); + expect(request.state == "Broken screen.\n\nSecond message.", + "user messages with input_text parts and strings join the state"); + expect(request.questions.size() == 3, "three questions"); + const DecisionQuestion& damaged = request.questions[0]; + expect(damaged.type == DecisionType::Noul && damaged.named && damaged.id == "damaged" && + damaged.options.size() == 2, + "a predicate is a noul"); + const DecisionQuestion& department = request.questions[1]; + expect(department.type == DecisionType::Choice && department.options.size() == 2 && + department.options[0].text == "returns" && department.options[0].description == "RMA" && + department.options[1].text == "true" && department.options[1].value == Json(true), + "choice values keep their type"); + const DecisionQuestion& severity = request.questions[2]; + expect(severity.type == DecisionType::Score && !severity.named && + severity.options[1].text == "broken", + "an unnamed score with labelled levels"); + + expect(decisions_refusal(R"({"input": "i", "questions": [{"type": "noul", "instructions": "i"}]})") + .param == "questions[0].type", + "System One's noul is not an OpenAI type"); + expect(decisions_refusal( + R"({"input": "i", "questions": [{"type": "choice", "instructions": "i", + "choices": [{"value": "a"}, {"value": "a"}]}]})") + .param == "questions[0].choices[1]", + "choice values are unique"); + expect(decisions_refusal( + R"({"input": [{"role": "system", "content": "s"}], + "questions": [{"type": "predicate", "instructions": "i"}]})") + .param == "input[0].role", + "only user messages"); + expect(decisions_refusal( + R"({"input": "i", "stream": true, "questions": [{"type": "predicate", "instructions": "i"}]})") + .param == "stream", + "unknown fields are refused"); + expect(decisions_refusal( + R"({"input": "i", "questions": [{"type": "predicate", "instructions": "i", + "levels": [{"label": "a"}, {"label": "b"}]}]})") + .param == "questions[0]", + "a predicate takes no levels"); +} + +void prompts() { + const DecisionRequest request = parse_systemone_request(RequestJson::parse(kSystemOne)); + const DecisionLabels team = decision_labels(request.questions[0]); + const DecisionLabels urgent = decision_labels(request.questions[1]); + const DecisionLabels tone = decision_labels(request.questions[2]); + expect(team.texts == std::vector({"A", "B"}) && + team.option == std::vector({0, 1}) && + tone.texts == std::vector({"0", "1", "2"}) && + tone.option == std::vector({0, 1, 2}), + "letters and digits label the options"); + expect(urgent.texts == std::vector({"yes", "Yes", "no", "No"}) && + urgent.option == std::vector({0, 0, 1, 1}), + "a noul reads yes and no in both cases"); + const GenerationRequest choice = make_decision_read(request, request.questions[0], {32, 33}); + expect(choice.messages.size() == 2 && choice.messages[0].role == ninfer::ChatRole::System && + choice.messages[0].content.at(0).text.find("one label only") != std::string::npos && + choice.messages[1].role == ninfer::ChatRole::User, + "the system turn asking for one label, then the user turn"); + const std::string& text = choice.messages[1].content.at(0).text; + expect(text.find("State:\n{\"ticket\"") == 0 && + text.find("Question: Which team?\nOptions:\nA. billing - payments and refunds\n" + "B. shipping\nReply with a single letter (A or B) and nothing else.") != + std::string::npos, + "the state, the question and the labelled options"); + expect(choice.enable_thinking == false && choice.max_tokens == 1 && choice.logprobs && + choice.readout_tokens == std::vector({32, 33}) && choice.graft && + choice.graft->empty() && choice.sampling.temperature == 0.0 && + choice.sampling.presence_penalty == 0.0, + "one greedy token with the labels read out, no thinking, no graft, no penalties"); + const GenerationRequest noul = make_decision_read(request, request.questions[1], {1, 2, 3, 4}); + expect(noul.messages[1].content.at(0).text.find( + "Question: Reply within the hour?\nyes: money is moving\n" + "Answer with one word, yes or no, in lower case.") != std::string::npos, + "a noul shows the described answers"); + const GenerationRequest score = make_decision_read(request, request.questions[2], {1, 2, 3}); + expect(score.messages[1].content.at(0).text.find( + "Levels, lowest first:\n0. calm\n1. annoyed\n2. furious\n" + "Reply with a single number (0, 1 or 2) and nothing else.") != std::string::npos, + "a score lists its levels under their numbers"); + const DecisionRequest with_context = parse_systemone_request(RequestJson::parse( + R"({"state": "s", "instructions": "ctx", + "questions": {"q": {"type": "noul", "instructions": "i"}, + "wide": {"type": "choice", "instructions": "i", + "criteria": {"a": null, "b": null, "c": null, "d": null, "e": null}}}})")); + expect(make_decision_read(with_context, with_context.questions[0], {1, 2, 3, 4}) + .messages[1] + .content.at(0) + .text.rfind("ctx\n\nState:\ns\n\n", 0) == 0, + "context goes ahead of the state"); + expect(make_decision_read(with_context, with_context.questions[1], {1, 2, 3, 4, 5}) + .messages[1] + .content.at(0) + .text.find("Reply with a single letter (A to E) and nothing else.") != + std::string::npos, + "many options are spanned"); +} + +void reading() { + const DecisionLabels three{.texts = {"A", "B", "C"}, .option = {0, 1, 2}}; + const std::vector labels{7, 8, 9}; + // Exact readout: 0.6, 0.2 and 0.1 of the vocabulary on the labels. + const DecisionAnswer exact = + read_decision(read({std::log(0.6), std::log(0.2), std::log(0.1)}, {{7, 0.6}, {8, 0.2}}), + three, labels); + expect(exact.exact && close_to(exact.label_mass, 0.9) && + close_to(exact.probabilities[0], 0.6 / 0.9) && + close_to(exact.probabilities[2], 0.1 / 0.9) && exact.argmax_is_label, + "softmax over the exact label log probabilities, their total as the label mass"); + const DecisionAnswer other = read_decision( + read({std::log(0.3), std::log(0.1), std::log(0.1)}, {{5, 0.5}, {7, 0.3}}), three, labels); + expect(!other.argmax_is_label && close_to(other.label_mass, 0.5), + "the model's favourite token outside the labels is reported"); + // No readout: the top alternatives, a missing label as likely as the least likely of them. + const DecisionAnswer top = + read_decision(read({}, {{7, 0.5}, {4, 0.2}, {9, 0.1}}), three, labels); + expect(!top.exact && close_to(top.label_mass, 0.7) && close_to(top.probabilities[0], 0.5 / 0.7) && + close_to(top.probabilities[1], 0.1 / 0.7) && close_to(top.probabilities[2], 0.1 / 0.7), + "without a readout the labels are read from the top alternatives"); + const DecisionAnswer short_readout = read_decision(read({0.0, 0.0}, {{7, 0.5}}), three, labels); + expect(!short_readout.exact, "a readout of another size is not this question's"); + // A noul's two cases add up per option. + const DecisionLabels noul{.texts = {"yes", "Yes", "no", "No"}, .option = {0, 0, 1, 1}}; + const std::vector noul_tokens{11, 12, 13, 14}; + const DecisionAnswer merged = read_decision( + read({std::log(0.5), std::log(0.2), std::log(0.1), std::log(0.1)}, {{7, 0.5}}), noul, + noul_tokens); + expect(merged.exact && merged.probabilities.size() == 2 && + close_to(merged.probabilities[0], 0.7 / 0.9) && + close_to(merged.probabilities[1], 0.2 / 0.9) && close_to(merged.label_mass, 0.9), + "yes and Yes answer yes, no and No answer no"); + bool refused = false; + try { + (void)read_decision(GenerationOutcome{}, three, labels); + } catch (const std::runtime_error&) { refused = true; } + expect(refused, "a read without an answer token is an error, not an answer"); +} + +void responses() { + const DecisionRequest request = parse_systemone_request(RequestJson::parse(kSystemOne)); + std::vector answers(3); + answers[0] = {.probabilities = {0.25, 0.75}, .label_mass = 0.8, .argmax_is_label = true}; + answers[1] = {.probabilities = {0.9, 0.1}, .label_mass = 0.5, .argmax_is_label = false}; + answers[2] = {.probabilities = {0.2, 0.5, 0.3}, + .label_mass = 1.0, + .argmax_is_label = true, + .exact = false}; + const Json body = Json::parse(make_decision_response(request, answers, 120, 3)); + expect(body["object"] == "structured_decision" && body["model"] == "m" && + body["id"].get().rfind("decision-", 0) == 0 && + body["usage"]["input_tokens"] == 120 && body["usage"]["output_tokens"] == 3, + "a structured decision with usage"); + const Json& team = body["answers"]["team"]; + expect(team["type"] == "choice" && team["choice"] == "shipping" && + close_to(team["probabilities"]["billing"].get(), 0.25) && + close_to(team["confidence"].get(), 0.6), + "the choice, its probabilities and the winner's share of the vocabulary"); + const Json& urgent = body["answers"]["urgent"]; + expect(urgent["type"] == "noul" && close_to(urgent["noul"].get(), 0.9) && + close_to(urgent["probabilities"]["no"].get(), 0.1) && + close_to(urgent["confidence"].get(), 0.45), + "a noul's probability of yes"); + const Json& tone = body["answers"]["tone"]; + expect(tone["type"] == "score" && close_to(tone["score"].get(), 1.1) && + tone["legend"]["2"] == "furious" && + close_to(tone["probabilities"]["1"].get(), 0.5) && + close_to(tone["confidence"].get(), 0.5), + "a score's expected level, legend and probabilities by level index"); + expect(close_to(body["diagnostics"]["team"]["label_mass"].get(), 0.8) && + body["diagnostics"]["urgent"]["argmax_is_label"] == false && + body["diagnostics"]["team"]["readout"] == "exact" && + body["diagnostics"]["tone"]["readout"] == "top_k", + "diagnostics per question"); + + const DecisionRequest openai = parse_decisions_request(RequestJson::parse(kOpenAI)); + std::vector openai_answers(3); + openai_answers[0] = {.probabilities = {0.3, 0.7}, .label_mass = 1.0}; + openai_answers[1] = {.probabilities = {0.4, 0.6}, .label_mass = 0.5}; + openai_answers[2] = {.probabilities = {0.8, 0.2}, .label_mass = 1.0}; + const Json decision = Json::parse(make_decision_response(openai, openai_answers, 40, 3)); + expect(decision["object"] == "decision" && decision["answers"].is_array() && + decision["answers"].size() == 3 && decision["usage"]["total_tokens"] == 43, + "an OpenAI decision with answers in question order"); + const Json& damaged = decision["answers"][0]; + expect(damaged["type"] == "predicate" && damaged["name"] == "damaged" && + close_to(damaged["probability"].get(), 0.3), + "a predicate's probability"); + const Json& department = decision["answers"][1]; + expect(department["type"] == "choice" && department["choice"] == true && + department["probabilities"][0]["value"] == "returns" && + close_to(department["probabilities"][1]["probability"].get(), 0.6) && + close_to(department["confidence"].get(), 0.3), + "a choice echoes its values with their types"); + const Json& severity = decision["answers"][2]; + expect(severity["type"] == "score" && severity["name"].is_null() && + close_to(severity["score"].get(), 0.2) && + severity["probabilities"][1]["label"] == "broken" && + severity["probabilities"][1]["value"] == 1, + "an unnamed score with labelled, indexed levels"); +} + +} // namespace + +int main() { + try { + systemone_parsing(); + decisions_parsing(); + prompts(); + reading(); + responses(); + } catch (const std::exception& error) { + std::cerr << "error: " << error.what() << '\n'; + return 1; + } + if (failures != 0) { + std::cerr << failures << " expectation(s) failed\n"; + return 1; + } + std::cout << "systemone: ok\n"; + return 0; +}