vllm.entrypoints.generate.structured_decisions.strategies

Read strategies: how a model is asked for label probabilities.

Only models in READ_STRATEGIES are currently supported.

Classes:

• DiffusionGemmaCanvasStrategy –

  DiffusionGemma. Each question is one read-only request with one denoising

• JointSchemaStrategy –

  Clef-Omni. The whole request is one pooling read: the state, then every

• LiquidStrategy –

  Liquid decision models (d1). The prompt and the label tokens are the

• NextTokenStrategy –

  Autoregressive models. Each question is one request: the state, then the

• ReadStrategy –

Functions:

• cap_pixels –

  Shrink image to at most max_pixels pixels (bicubic).

• prompt_tail –

  The prompt's ids after its last added token.

• reply_label_ids –

  The prompt's ids after its last added token, and the token each of

• select_read_strategy –

  Raise ValueError when the model cannot serve structured decisions.

• type_label_ids –

  The reply's first token for each label of every registered question

DiffusionGemmaCanvasStrategy

Bases: NextTokenStrategy

DiffusionGemma. Each question is one read-only request with one denoising step.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

class DiffusionGemmaCanvasStrategy(NextTokenStrategy):
    """DiffusionGemma. Each question is one read-only request with one denoising
    step."""

    THOUGHT = "<|channel>thought\n<channel|>"
    END_OF_TURN = "<turn|>"
    CANVAS_STEP = 16

    def __init__(self, context: ReadContext):
        ReadStrategy.__init__(self, context)
        tokenizer = self._tokenizer()
        self.tail = prompt_tail(tokenizer, self._generation_prompt())
        self.thought = tokenizer.encode(self.THOUGHT, add_special_tokens=False)
        self.label_ids = type_label_ids(tokenizer, self.thought)
        end = tokenizer.encode(self.END_OF_TURN, add_special_tokens=False)
        self.pad = tokenizer.pad_token_id
        if self.pad is None or len(end) != 1:
            raise ValueError(
                "Structured decisions need the model's "
                f"{self.END_OF_TURN!r} and pad tokens"
            )
        self.end = end[0]
        vllm_config = context.engine_client.vllm_config
        served = (
            vllm_config.diffusion_config.canvas_length
            if vllm_config.diffusion_config
            and vllm_config.diffusion_config.canvas_length
            else vllm_config.model_config.hf_config.canvas_length
        )
        self.width = self.CANVAS_STEP
        if self.width > served:
            raise ValueError(
                f"Structured decisions need a canvas of {self.width}, "
                f"and the served canvas is {served}"
            )
        self.vocab_size = vllm_config.model_config.get_vocab_size()
        self.max_model_len = vllm_config.model_config.max_model_len

    def _sampling_params(
        self, label_ids: list[int], prompt_ids: list[int]
    ) -> SamplingParams:
        # The noise at the label slot is drawn from the prompt, so a repeated
        # read gets the same canvas.
        rng = random.Random(hashlib.sha256(json.dumps(prompt_ids).encode()).digest())
        canvas = [rng.randrange(self.vocab_size), self.end]
        canvas += [self.pad] * (self.width - len(canvas))
        return SamplingParams(
            max_tokens=2,
            logprob_token_ids=label_ids,
            extra_args={
                "diffusion_seed_canvas": canvas,
                "diffusion_canvas_length": self.width,
                "diffusion_max_steps": 1,
                "diffusion_read_only": True,
            },
        )

    def _read_input(
        self, engine_input: EngineInput, prompt_ids: list[int]
    ) -> EngineInput:
        extra = len(self.thought) + self.width
        if len(prompt_ids) + extra > self.max_model_len:
            raise StructuredDecisionError(
                f"the prompt has {len(prompt_ids)} tokens and a canvas read adds "
                f"{extra}, more than max_model_len={self.max_model_len}"
            )
        salt = engine_input.get("cache_salt")
        # The thought block _must_ appear outside of the canvas (far worse results
        # otherwise)
        return tokens_input(
            prompt_ids + self.thought,
            cache_salt=salt if isinstance(salt, str) else None,
        )

JointSchemaStrategy

Bases: ReadStrategy

Clef-Omni. The whole request is one pooling read: the state, then every question with its options, as the model was trained to read them. The model's joint schema head scores each option, so a read has no labels.

Mirrors encode_record in the checkpoint's joint_schema_model.py. Each piece is tokenized on its own, so the spans of the questions and the options are known. The spans count from the end of the prompt, so they hold after the media placeholders, which come before the state, expand.

Methods:

• encode –

  The prompt, the schema for the pooler, and each question's option

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

class JointSchemaStrategy(ReadStrategy):
    """Clef-Omni. The whole request is one pooling read: the state, then every
    question with its options, as the model was trained to read them. The
    model's joint schema head scores each option, so a read has no labels.

    Mirrors ``encode_record`` in the checkpoint's ``joint_schema_model.py``.
    Each piece is tokenized on its own, so the spans of the questions and the
    options are known. The spans count from the end of the prompt, so they
    hold after the media placeholders, which come before the state, expand."""

    task = "token_classify"
    modalities = frozenset({"image", "audio", "video"})

    SYSTEM_PROMPT = (
        "Read the complete state and schema. Decide every field jointly. Each "
        "answer must be exactly one of that field's allowed options."
    )
    #: The reference's default bound on a prompt, in tokens.
    MAX_LENGTH = 64000
    TYPE_IDS = {"noul": 0, "choice": 1, "score": 2}
    NOUL_CRITERIA = {
        "true": "The proposition is true or the answer is yes.",
        "false": "The proposition is false or the answer is no.",
    }
    PLACEHOLDERS = {
        "image": "<|vision_start|><|image_pad|><|vision_end|>\n",
        "audio": "<|audio_start|><|audio_pad|><|audio_end|>\n",
        "video": "<|vision_start|><|video_pad|><|vision_end|>\n",
    }
    IMAGE_SIZE = {"shortest_edge": 65_536, "longest_edge": 1_048_576}
    #: The reference samples videos at 2 fps; the "clef_omni" video backend
    #: reproduces the reference's sampling, pixel cap and frame count cap.
    VIDEO_FPS = 2

    def __init__(self, context: ReadContext):
        super().__init__(context)
        self.tokenizer = context.online_renderer.renderer.get_tokenizer()
        self.max_length = min(
            self.MAX_LENGTH, context.engine_client.model_config.max_model_len
        )
        self.prefix = self._tokens(
            f"<|im_start|>system\n{self.SYSTEM_PROMPT}<|im_end|>\n"
            "<|im_start|>user\nSTATE:\n"
        )
        self.suffix = self._tokens(
            "\n<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
            "JOINT SCHEMA DECISIONS:"
        )

    def limits(self) -> DecisionLimits:
        return DecisionLimits(max_questions=64, max_options=len(LABELS))

    def render_state(self, state: Any) -> str:
        return self._render(state)

    def num_requests(self, questions: list[Question]) -> int:
        return 1

    @staticmethod
    def _render(value: Any) -> str:
        if isinstance(value, str):
            return value
        return json.dumps(
            value, ensure_ascii=False, separators=(",", ":"), sort_keys=True
        )

    def _tokens(self, text: str) -> list[int]:
        return self.tokenizer.encode(text, add_special_tokens=False)

    async def _fetch(
        self, urls: Sequence[str], media_io: MediaIO[_M], timeout: int
    ) -> list[_M]:
        return await asyncio.gather(
            *(
                self._connector.load_from_url_async(
                    url, media_io, fetch_timeout=timeout
                )
                for url in urls
            )
        )

    async def _soundtrack(
        self, video_url: str
    ) -> tuple[np.ndarray, int | float] | None:
        """The video's soundtrack, or None when it has no decodable one."""
        try:
            return await self._connector.fetch_audio_async(video_url)
        except ValueError as e:
            logger.debug("no decodable soundtrack in %s: %s", video_url, e)
            return None

    async def _render_media(
        self, media: Mapping[str, Sequence[str]], cache_salt: str | None
    ) -> MultiModalInput:
        """The system prompt and the media, with the media's placeholders
        expanded. A video's soundtrack is heard with it when every video has
        one."""
        connector = self._connector
        video_io = VideoMediaIO(
            ImageMediaIO(),
            num_frames=-1,
            fps=self.VIDEO_FPS,
            video_backend="clef_omni",
        )
        try:
            images, audio, videos, tracks = await asyncio.gather(
                asyncio.gather(
                    *map(connector.fetch_image_async, media.get("image", ()))
                ),
                asyncio.gather(
                    *map(connector.fetch_audio_async, media.get("audio", ()))
                ),
                self._fetch(
                    media.get("video", ()), video_io, envs.VLLM_VIDEO_FETCH_TIMEOUT
                ),
                asyncio.gather(*map(self._soundtrack, media.get("video", ()))),
            )
        except ValueError as e:
            raise StructuredDecisionError(f"media: {e}") from e

        audio_in_video = bool(videos) and all(t is not None for t in tracks)
        if audio_in_video and audio:
            raise StructuredDecisionError(
                "audio clips cannot be read with videos that have soundtracks"
            )
        text = (
            self.PLACEHOLDERS["image"] * len(images)
            + self.PLACEHOLDERS["audio"] * len(audio)
            + self.PLACEHOLDERS["video"] * len(videos)
        )
        if audio_in_video:
            audio = [track for track in tracks if track is not None]
        mm_data: dict[str, Any] = {}
        if images:
            mm_data["image"] = images
        if audio:
            mm_data["audio"] = audio
        if videos:
            mm_data["video"] = videos
        mm_kwargs: dict[str, Any] = {"images_kwargs": {"size": self.IMAGE_SIZE}}
        if videos:
            mm_kwargs["use_audio_in_video"] = audio_in_video
        prompt = TokensPrompt(
            prompt_token_ids=self.prefix + self._tokens(text),
            multi_modal_data=mm_data,
            mm_processor_kwargs=mm_kwargs,
        )
        if cache_salt is not None:
            prompt["cache_salt"] = cache_salt
        renderer = self.context.online_renderer.renderer
        try:
            engine_input = await renderer.process_for_engine_async(prompt, time.time())
        except ValueError as e:
            raise StructuredDecisionError(f"media: {e}") from e
        assert engine_input["type"] == "multimodal"
        return engine_input

    def _options(self, q: Question) -> list[tuple[int, str, Any]]:
        """Each option's index in ``q.options``, id and description, in the
        model's order: choices sorted by name, noul true then false, score
        levels by index."""
        if q.type.name == "noul":
            return [
                (
                    i,
                    key,
                    self.NOUL_CRITERIA[key] if o.description is None else o.description,
                )
                for i, (key, o) in enumerate(zip(("true", "false"), q.options))
            ]
        if q.type.name == "choice":
            return sorted(
                ((i, o.name, o.description) for i, o in enumerate(q.options)),
                key=lambda option: option[1],
            )
        if q.type.name == "score":
            return [(i, str(i), o.name) for i, o in enumerate(q.options)]
        raise StructuredDecisionError(
            f"question {q.id!r}: this model does not answer {q.type.name!r} questions"
        )

    def encode(
        self, questions: list[Question], state: str, prefix: list[int] | None = None
    ) -> tuple[list[int], dict[str, list[Any]], list[list[int]]]:
        """The prompt, the schema for the pooler, and each question's option
        indices in the order of the output. ``prefix`` is what precedes the
        state: the system prompt, and the media if there are any."""
        prefix = self.prefix if prefix is None else prefix
        schema_ids = self._tokens("\n\nSCHEMA FIELDS:\n")
        question_spans, option_spans, orders = [], [], []
        for n, q in enumerate(questions, start=1):
            schema_ids += self._tokens(
                f"\nFIELD {n}\nID: {q.id}\nTYPE: {q.type.name}\nINSTRUCTION: "
            )
            start = len(schema_ids)
            schema_ids += self._tokens(q.instructions or q.id)
            question_spans.append([start, len(schema_ids)])
            schema_ids += self._tokens("\nALLOWED OPTIONS:\n")
            options = self._options(q)
            for m, (_, option_id, description) in enumerate(options, start=1):
                schema_ids += self._tokens(f"OPTION {m}: ")
                start = len(schema_ids)
                semantics = {"option_id": option_id}
                if description is not None:
                    semantics["description"] = description
                schema_ids += self._tokens(self._render(semantics))
                option_spans.append([start, len(schema_ids)])
                schema_ids += self._tokens("\n")
            schema_ids += self._tokens("END FIELD\n")
            orders.append([i for i, _, _ in options])

        fixed = len(prefix) + len(schema_ids) + len(self.suffix)
        if fixed > self.max_length:
            raise StructuredDecisionError(
                f"the questions take {fixed} tokens, more than the "
                f"{self.max_length} this model reads"
            )
        state_ids = self._tokens(state)[: self.max_length - fixed]
        prompt = prefix + state_ids + schema_ids + self.suffix
        # The schema ends len(self.suffix) tokens before the prompt does.
        end = len(schema_ids) + len(self.suffix)
        schema: dict[str, list[Any]] = {
            "question_types": [self.TYPE_IDS[q.type.name] for q in questions],
            "question_spans": [[s - end, e - end] for s, e in question_spans],
            "option_spans": [[s - end, e - end] for s, e in option_spans],
            "option_counts": [len(order) for order in orders],
        }
        return prompt, schema, orders

    async def read(
        self,
        questions: list[Question],
        instructions: str | None,
        state: str,
        *,
        request_id: str,
        chat_template_kwargs: dict[str, Any] | None,
        lora_request: LoRARequest | None,
        priority: int,
        cache_salt: str | None = None,
        trace_headers: Mapping[str, str] | None = None,
        media: Mapping[str, Sequence[str]] | None = None,
    ) -> list[QuestionRead]:
        if instructions:
            raise StructuredDecisionError(
                "this model reads a fixed prompt, so it takes no instructions"
            )
        engine_input: EngineInput
        if media:
            engine_input = await self._render_media(media, cache_salt)
            prompt, schema, orders = self.encode(
                questions, state, engine_input["prompt_token_ids"]
            )
            engine_input["prompt_token_ids"] = prompt
        else:
            prompt, schema, orders = self.encode(questions, state)
            engine_input = tokens_input(prompt, cache_salt=cache_salt)
        result = None
        async for result in self.context.engine_client.encode(
            engine_input,
            PoolingParams(task="token_classify", extra_kwargs={"joint_schema": schema}),
            f"{request_id}-0",
            lora_request=lora_request,
            trace_headers=trace_headers,
            priority=priority,
        ):
            pass
        if result is None:
            raise ValueError("Failed to score the questions")
        if result.error is not None:
            raise ValueError(f"Failed to score the questions: {result.error.message}")
        probs = result.outputs.data.tolist()
        if len(probs) != sum(schema["option_counts"]):
            raise ValueError(
                f"Expected {sum(schema['option_counts'])} option scores, "
                f"got {len(probs)}"
            )

        reads, offset = [], 0
        for n, order in enumerate(orders):
            q_probs = [0.0] * len(order)
            for i, p in zip(order, probs[offset : offset + len(order)]):
                q_probs[i] = p
            offset += len(order)
            reads.append(
                QuestionRead(
                    probs=q_probs,
                    label_mass=1.0,
                    argmax_is_label=True,
                    # One forward pass reads the prompt for every question.
                    input_tokens=len(prompt) if n == 0 else 0,
                    output_tokens=0,
                )
            )
        return reads

_options(q)

Each option's index in q.options, id and description, in the model's order: choices sorted by name, noul true then false, score levels by index.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def _options(self, q: Question) -> list[tuple[int, str, Any]]:
    """Each option's index in ``q.options``, id and description, in the
    model's order: choices sorted by name, noul true then false, score
    levels by index."""
    if q.type.name == "noul":
        return [
            (
                i,
                key,
                self.NOUL_CRITERIA[key] if o.description is None else o.description,
            )
            for i, (key, o) in enumerate(zip(("true", "false"), q.options))
        ]
    if q.type.name == "choice":
        return sorted(
            ((i, o.name, o.description) for i, o in enumerate(q.options)),
            key=lambda option: option[1],
        )
    if q.type.name == "score":
        return [(i, str(i), o.name) for i, o in enumerate(q.options)]
    raise StructuredDecisionError(
        f"question {q.id!r}: this model does not answer {q.type.name!r} questions"
    )

_render_media(media, cache_salt) async

The system prompt and the media, with the media's placeholders expanded. A video's soundtrack is heard with it when every video has one.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

async def _render_media(
    self, media: Mapping[str, Sequence[str]], cache_salt: str | None
) -> MultiModalInput:
    """The system prompt and the media, with the media's placeholders
    expanded. A video's soundtrack is heard with it when every video has
    one."""
    connector = self._connector
    video_io = VideoMediaIO(
        ImageMediaIO(),
        num_frames=-1,
        fps=self.VIDEO_FPS,
        video_backend="clef_omni",
    )
    try:
        images, audio, videos, tracks = await asyncio.gather(
            asyncio.gather(
                *map(connector.fetch_image_async, media.get("image", ()))
            ),
            asyncio.gather(
                *map(connector.fetch_audio_async, media.get("audio", ()))
            ),
            self._fetch(
                media.get("video", ()), video_io, envs.VLLM_VIDEO_FETCH_TIMEOUT
            ),
            asyncio.gather(*map(self._soundtrack, media.get("video", ()))),
        )
    except ValueError as e:
        raise StructuredDecisionError(f"media: {e}") from e

    audio_in_video = bool(videos) and all(t is not None for t in tracks)
    if audio_in_video and audio:
        raise StructuredDecisionError(
            "audio clips cannot be read with videos that have soundtracks"
        )
    text = (
        self.PLACEHOLDERS["image"] * len(images)
        + self.PLACEHOLDERS["audio"] * len(audio)
        + self.PLACEHOLDERS["video"] * len(videos)
    )
    if audio_in_video:
        audio = [track for track in tracks if track is not None]
    mm_data: dict[str, Any] = {}
    if images:
        mm_data["image"] = images
    if audio:
        mm_data["audio"] = audio
    if videos:
        mm_data["video"] = videos
    mm_kwargs: dict[str, Any] = {"images_kwargs": {"size": self.IMAGE_SIZE}}
    if videos:
        mm_kwargs["use_audio_in_video"] = audio_in_video
    prompt = TokensPrompt(
        prompt_token_ids=self.prefix + self._tokens(text),
        multi_modal_data=mm_data,
        mm_processor_kwargs=mm_kwargs,
    )
    if cache_salt is not None:
        prompt["cache_salt"] = cache_salt
    renderer = self.context.online_renderer.renderer
    try:
        engine_input = await renderer.process_for_engine_async(prompt, time.time())
    except ValueError as e:
        raise StructuredDecisionError(f"media: {e}") from e
    assert engine_input["type"] == "multimodal"
    return engine_input

_soundtrack(video_url) async

The video's soundtrack, or None when it has no decodable one.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

async def _soundtrack(
    self, video_url: str
) -> tuple[np.ndarray, int | float] | None:
    """The video's soundtrack, or None when it has no decodable one."""
    try:
        return await self._connector.fetch_audio_async(video_url)
    except ValueError as e:
        logger.debug("no decodable soundtrack in %s: %s", video_url, e)
        return None

encode(questions, state, prefix=None)

The prompt, the schema for the pooler, and each question's option indices in the order of the output. prefix is what precedes the state: the system prompt, and the media if there are any.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def encode(
    self, questions: list[Question], state: str, prefix: list[int] | None = None
) -> tuple[list[int], dict[str, list[Any]], list[list[int]]]:
    """The prompt, the schema for the pooler, and each question's option
    indices in the order of the output. ``prefix`` is what precedes the
    state: the system prompt, and the media if there are any."""
    prefix = self.prefix if prefix is None else prefix
    schema_ids = self._tokens("\n\nSCHEMA FIELDS:\n")
    question_spans, option_spans, orders = [], [], []
    for n, q in enumerate(questions, start=1):
        schema_ids += self._tokens(
            f"\nFIELD {n}\nID: {q.id}\nTYPE: {q.type.name}\nINSTRUCTION: "
        )
        start = len(schema_ids)
        schema_ids += self._tokens(q.instructions or q.id)
        question_spans.append([start, len(schema_ids)])
        schema_ids += self._tokens("\nALLOWED OPTIONS:\n")
        options = self._options(q)
        for m, (_, option_id, description) in enumerate(options, start=1):
            schema_ids += self._tokens(f"OPTION {m}: ")
            start = len(schema_ids)
            semantics = {"option_id": option_id}
            if description is not None:
                semantics["description"] = description
            schema_ids += self._tokens(self._render(semantics))
            option_spans.append([start, len(schema_ids)])
            schema_ids += self._tokens("\n")
        schema_ids += self._tokens("END FIELD\n")
        orders.append([i for i, _, _ in options])

    fixed = len(prefix) + len(schema_ids) + len(self.suffix)
    if fixed > self.max_length:
        raise StructuredDecisionError(
            f"the questions take {fixed} tokens, more than the "
            f"{self.max_length} this model reads"
        )
    state_ids = self._tokens(state)[: self.max_length - fixed]
    prompt = prefix + state_ids + schema_ids + self.suffix
    # The schema ends len(self.suffix) tokens before the prompt does.
    end = len(schema_ids) + len(self.suffix)
    schema: dict[str, list[Any]] = {
        "question_types": [self.TYPE_IDS[q.type.name] for q in questions],
        "question_spans": [[s - end, e - end] for s, e in question_spans],
        "option_spans": [[s - end, e - end] for s, e in option_spans],
        "option_counts": [len(order) for order in orders],
    }
    return prompt, schema, orders

LiquidStrategy

Bases: NextTokenStrategy

Liquid decision models (d1). The prompt and the label tokens are the ones the checkpoint was trained with, from the prompt.py it ships.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

class LiquidStrategy(NextTokenStrategy):
    """Liquid decision models (d1). The prompt and the label tokens are the
    ones the checkpoint was trained with, from the `prompt.py` it ships."""

    modalities = frozenset({"image"})
    # runner.py bounds every picture at 1024 x 1024 pixels before the processor.
    max_image_pixels = 1024 * 1024

    def render_state(self, state: Any) -> str:
        if state is None:
            return ""
        if isinstance(state, str):
            return state
        return json.dumps(state, ensure_ascii=False, indent=2)

    def _single_tokens(self, texts: Sequence[str]) -> list[int]:
        ids: list[int] = []
        for text in texts:
            encoded = self._tokenizer().encode(text, add_special_tokens=False)
            if len(encoded) == 1 and encoded[0] not in ids:
                ids.append(encoded[0])
        return ids

    @staticmethod
    def _codes(question: Question) -> list[str]:
        """A choice's codes are A to Z, or the option names when those are
        already single letters."""
        names = [o.name.strip() for o in question.options]
        if all(len(n) == 1 and n.isalpha() for n in names):
            return names
        return list(question.labels)

    def _label_groups(self, question: Question) -> list[list[int]]:
        if question.type.name == "noul":
            forms: list[Sequence[str]] = [("yes", "Yes", "YES"), ("no", "No", "NO")]
        elif question.type.name == "choice":
            forms = [(code, f" {code}") for code in self._codes(question)]
        else:
            forms = [(label,) for label in question.labels]
        groups = [self._single_tokens(f) for f in forms]
        seen: set[int] = set()
        for (label, *_), ids in zip(forms, groups):
            if not ids or self._single_tokens([label]) != ids[:1] or seen & set(ids):
                raise StructuredDecisionError(
                    f"question {question.id!r}: label {label!r} is not one "
                    "distinct token for this model"
                )
            seen.update(ids)
        return groups

    def _content(self, state: str, question: Question) -> str:
        q = question
        if q.type.name == "choice":
            lines = "\n".join(
                f"{code} {o.description or o.name.replace('_', ' ')}"
                for code, o in zip(self._codes(q), q.options)
            )
            ask = (
                f"{q.instructions}\n\nOptions:\n{lines}\n\n"
                "Reply with the option code only."
            )
        elif q.type.name == "noul":
            yes, no = (o.description for o in q.options)
            sides = f"\nYes: {yes}\nNo: {no}" if yes or no else ""
            ask = f"{q.instructions}{sides}\n\nReply with yes or no only."
        elif q.type.name == "score":
            legend = "\n".join(
                f"{label} {o.name}" for label, o in zip(q.labels, q.options)
            )
            ask = (
                f"{q.instructions}\n\n{legend}\n\n"
                f"Reply with a single digit 0-{len(q.options) - 1} only."
            )
        else:
            raise StructuredDecisionError(
                f"question {q.id!r}: this model does not answer {q.type.name!r} "
                "questions"
            )
        # A null state leaves the images, or the question alone, as the prompt.
        return f"{state}\n\n\nQUESTION:\n{ask}" if state else ask

_codes(question) staticmethod

A choice's codes are A to Z, or the option names when those are already single letters.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

@staticmethod
def _codes(question: Question) -> list[str]:
    """A choice's codes are A to Z, or the option names when those are
    already single letters."""
    names = [o.name.strip() for o in question.options]
    if all(len(n) == 1 and n.isalpha() for n in names):
        return names
    return list(question.labels)

NextTokenStrategy

Bases: ReadStrategy

Autoregressive models. Each question is one request: the state, then the question with its labeled options, and the logprobs of the label tokens as the reply's first token. The requests share the state. With prefix caching, it is prefilled once.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

class NextTokenStrategy(ReadStrategy):
    """Autoregressive models. Each question is one request: the state, then the
    question with its labeled options, and the logprobs of the label tokens as
    the reply's first token. The requests share the state. With prefix caching,
    it is prefilled once."""

    #: Images are shrunk to at most this many pixels before the processor.
    max_image_pixels: ClassVar[int | None] = None

    def __init__(self, context: ReadContext):
        super().__init__(context)
        # Every prompt ends with the same generation prompt, so a label's token
        # is the same for every question.
        prompt = self._generation_prompt()
        self.tail = prompt_tail(self._tokenizer(), prompt)
        self.label_ids = type_label_ids(self._tokenizer(), prompt)

    def _tokenizer(self) -> TokenizerLike:
        return self.context.online_renderer.renderer.get_tokenizer()

    def _generation_prompt(self) -> list[int]:
        """A chat prompt's ids up to the generation prompt, with thinking off."""
        context, tokenizer = self.context, self._tokenizer()
        probe = tokenizer.apply_chat_template(
            [{"role": "user", "content": "x"}],
            **{
                **context.default_chat_template_kwargs,
                "chat_template": context.chat_template,
                "add_generation_prompt": True,
                "tokenize": True,
                "return_dict": False,
                "enable_thinking": False,
            },
        )
        if isinstance(probe, str):
            probe = tokenizer.encode(probe, add_special_tokens=False)
        return list(probe)

    def _sampling_params(
        self, label_ids: list[int], prompt_ids: list[int]
    ) -> SamplingParams:
        return SamplingParams(
            max_tokens=1, temperature=0.0, logprob_token_ids=label_ids
        )

    def _read_input(
        self, engine_input: EngineInput, prompt_ids: list[int]
    ) -> EngineInput:
        """The engine input for a read of the rendered prompt."""
        return engine_input

    def _label_groups(self, question: Question) -> list[list[int]]:
        """The tokens that count for each label. A label scores its best one."""
        return [[self.label_ids[label]] for label in question.labels]

    def _content(self, state: str, question: Question) -> str:
        """The user turn: the state, then the question."""
        return f"{state}\n\n{question.type.prompt(question)}"

    async def _image_parts(self, urls: Sequence[str]) -> list[dict[str, Any]]:
        """The images as chat content parts, fetched once so that every
        question reads the same pictures."""
        if not urls:
            return []
        try:
            images = await asyncio.gather(*map(self._connector.fetch_image_async, urls))
        except ValueError as e:
            raise StructuredDecisionError(f"media: {e}") from e
        if self.max_image_pixels is not None:
            images = [cap_pixels(image, self.max_image_pixels) for image in images]
        return [{"type": "image_pil", "image_pil": image} for image in images]

    def limits(self) -> DecisionLimits:
        return DecisionLimits(max_questions=64, max_options=len(LABELS))

    def _read_request(
        self, chat_template_kwargs: dict[str, Any] | None, cache_salt: str | None = None
    ) -> ReadPromptRequest:
        if (chat_template_kwargs or {}).get("enable_thinking"):
            raise StructuredDecisionError(
                "a read is the reply's first token, so thinking must be off"
            )
        return ReadPromptRequest(
            chat_template_kwargs=chat_template_kwargs, cache_salt=cache_salt
        )

    async def _render(
        self, read_request: ReadPromptRequest, messages: list[dict[str, Any]]
    ) -> tuple[EngineInput, list[int]]:
        ctx = self.context
        _, (engine_input,) = await ctx.online_renderer.preprocess_chat(
            read_request,
            messages,
            default_template=ctx.chat_template,
            default_template_content_format=ctx.chat_template_content_format,
            default_template_kwargs=ctx.default_chat_template_kwargs,
            # A read that fails before it reaches the engine must not leave the
            # sender cache believing the engine has the images.
            skip_mm_cache=True,
        )
        prompt_ids = extract_prompt_components(
            ctx.engine_client.model_config, engine_input
        ).token_ids
        return engine_input, list(prompt_ids or [])

    async def read(
        self,
        questions: list[Question],
        instructions: str | None,
        state: str,
        *,
        request_id: str,
        chat_template_kwargs: dict[str, Any] | None,
        lora_request: LoRARequest | None,
        priority: int,
        cache_salt: str | None = None,
        trace_headers: Mapping[str, str] | None = None,
        media: Mapping[str, Sequence[str]] | None = None,
    ) -> list[QuestionRead]:
        ctx = self.context
        read_request = self._read_request(chat_template_kwargs, cache_salt)
        image_parts = await self._image_parts((media or {}).get("image", ()))

        slots, engine_inputs, params = [], [], []
        for q in questions:
            content: Any = self._content(state, q)
            if image_parts:
                content = [*image_parts, {"type": "text", "text": content}]
            messages = [{"role": "user", "content": content}]
            if instructions:
                messages.insert(0, {"role": "system", "content": instructions})
            engine_input, prompt_ids = await self._render(read_request, messages)
            if prompt_ids[len(prompt_ids) - len(self.tail) :] != self.tail:
                raise StructuredDecisionError(
                    "these chat options end the prompt differently, so the "
                    "labels' tokens are unknown"
                )
            slots.append(self._label_groups(q))
            engine_inputs.append(self._read_input(engine_input, prompt_ids))
            params.append(
                self._sampling_params([i for ids in slots[-1] for i in ids], prompt_ids)
            )

        label_reads = await next_token_label_reads(
            ctx.engine_client,
            engine_inputs,
            params,
            request_id,
            lora_request=lora_request,
            trace_headers=trace_headers,
            priority=priority,
        )

        reads = []
        for groups, label_read in zip(slots, label_reads):
            output = label_read.result.outputs[0]
            logprobs = iter(label_read.logprobs)
            reads.append(
                QuestionRead(
                    probs=label_softmax(
                        [max(next(logprobs) for _ in g) for g in groups]
                    ),
                    label_mass=sum(math.exp(lp) for lp in label_read.logprobs),
                    argmax_is_label=bool(output.token_ids)
                    and any(output.token_ids[0] in ids for ids in groups),
                    input_tokens=len(label_read.result.prompt_token_ids or ()),
                    output_tokens=len(output.token_ids),
                )
            )
        return reads

_content(state, question)

The user turn: the state, then the question.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def _content(self, state: str, question: Question) -> str:
    """The user turn: the state, then the question."""
    return f"{state}\n\n{question.type.prompt(question)}"

_generation_prompt()

A chat prompt's ids up to the generation prompt, with thinking off.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def _generation_prompt(self) -> list[int]:
    """A chat prompt's ids up to the generation prompt, with thinking off."""
    context, tokenizer = self.context, self._tokenizer()
    probe = tokenizer.apply_chat_template(
        [{"role": "user", "content": "x"}],
        **{
            **context.default_chat_template_kwargs,
            "chat_template": context.chat_template,
            "add_generation_prompt": True,
            "tokenize": True,
            "return_dict": False,
            "enable_thinking": False,
        },
    )
    if isinstance(probe, str):
        probe = tokenizer.encode(probe, add_special_tokens=False)
    return list(probe)

_image_parts(urls) async

The images as chat content parts, fetched once so that every question reads the same pictures.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

async def _image_parts(self, urls: Sequence[str]) -> list[dict[str, Any]]:
    """The images as chat content parts, fetched once so that every
    question reads the same pictures."""
    if not urls:
        return []
    try:
        images = await asyncio.gather(*map(self._connector.fetch_image_async, urls))
    except ValueError as e:
        raise StructuredDecisionError(f"media: {e}") from e
    if self.max_image_pixels is not None:
        images = [cap_pixels(image, self.max_image_pixels) for image in images]
    return [{"type": "image_pil", "image_pil": image} for image in images]

_label_groups(question)

The tokens that count for each label. A label scores its best one.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def _label_groups(self, question: Question) -> list[list[int]]:
    """The tokens that count for each label. A label scores its best one."""
    return [[self.label_ids[label]] for label in question.labels]

_read_input(engine_input, prompt_ids)

The engine input for a read of the rendered prompt.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def _read_input(
    self, engine_input: EngineInput, prompt_ids: list[int]
) -> EngineInput:
    """The engine input for a read of the rendered prompt."""
    return engine_input

ReadStrategy

Bases: ABC

Methods:

• num_requests –

  The engine requests a decision takes.

• read –

  One read per question, in order.

• render_state –

  The request's state as the model reads it.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

class ReadStrategy(ABC):
    #: The engine task reads run as.
    task: ClassVar[SupportedTask] = "generate"
    #: The modalities a request may attach media of.
    modalities: ClassVar[frozenset[str]] = frozenset()

    def __init__(self, context: ReadContext):
        self.context = context

    @cached_property
    def _connector(self) -> MediaConnector:
        model_config = self.context.engine_client.model_config
        return MEDIA_CONNECTOR_REGISTRY.load(
            envs.VLLM_MEDIA_CONNECTOR,
            media_io_kwargs=model_config.get_multimodal_config().media_io_kwargs,
            allowed_local_media_path=model_config.allowed_local_media_path,
            allowed_media_domains=model_config.allowed_media_domains,
        )

    @abstractmethod
    def limits(self) -> DecisionLimits: ...

    def render_state(self, state: Any) -> str:
        """The request's state as the model reads it."""
        return state_text(state)

    def num_requests(self, questions: list[Question]) -> int:
        """The engine requests a decision takes."""
        return len(questions)

    @abstractmethod
    async def read(
        self,
        questions: list[Question],
        instructions: str | None,
        state: str,
        *,
        request_id: str,
        chat_template_kwargs: dict[str, Any] | None,
        lora_request: LoRARequest | None,
        priority: int,
        cache_salt: str | None = None,
        trace_headers: Mapping[str, str] | None = None,
        media: Mapping[str, Sequence[str]] | None = None,
    ) -> list[QuestionRead]:
        """One read per question, in order."""

num_requests(questions)

The engine requests a decision takes.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def num_requests(self, questions: list[Question]) -> int:
    """The engine requests a decision takes."""
    return len(questions)

read(questions, instructions, state, *, request_id, chat_template_kwargs, lora_request, priority, cache_salt=None, trace_headers=None, media=None) abstractmethod async

One read per question, in order.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

@abstractmethod
async def read(
    self,
    questions: list[Question],
    instructions: str | None,
    state: str,
    *,
    request_id: str,
    chat_template_kwargs: dict[str, Any] | None,
    lora_request: LoRARequest | None,
    priority: int,
    cache_salt: str | None = None,
    trace_headers: Mapping[str, str] | None = None,
    media: Mapping[str, Sequence[str]] | None = None,
) -> list[QuestionRead]:
    """One read per question, in order."""

render_state(state)

The request's state as the model reads it.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def render_state(self, state: Any) -> str:
    """The request's state as the model reads it."""
    return state_text(state)

cap_pixels(image, max_pixels)

Shrink image to at most max_pixels pixels (bicubic).

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def cap_pixels(image: Image.Image, max_pixels: int) -> Image.Image:
    """Shrink ``image`` to at most ``max_pixels`` pixels (bicubic)."""
    w, h = image.size
    if w * h <= max_pixels:
        return image
    scale = math.sqrt(max_pixels / (w * h))
    size = (max(1, int(w * scale)), max(1, int(h * scale)))
    return image.resize(size, Image.Resampling.BICUBIC)

prompt_tail(tokenizer, prompt_ids)

The prompt's ids after its last added token.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def prompt_tail(tokenizer: TokenizerLike, prompt_ids: Sequence[int]) -> list[int]:
    """The prompt's ids after its last added token."""
    added = set(tokenizer.get_added_vocab().values())
    start = max((i + 1 for i, t in enumerate(prompt_ids) if t in added), default=0)
    return list(prompt_ids[start:])

reply_label_ids(tokenizer, prompt_ids, labels=LABELS)

The prompt's ids after its last added token, and the token each of labels adds after them as the first token of the reply. Raises ValueError if a label is not one distinct token there.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def reply_label_ids(
    tokenizer: TokenizerLike,
    prompt_ids: Sequence[int],
    labels: Sequence[str] = LABELS,
) -> tuple[list[int], list[int]]:
    """The prompt's ids after its last added token, and the token each of
    ``labels`` adds after them as the first token of the reply. Raises
    ValueError if a label is not one distinct token there."""
    tail = prompt_tail(tokenizer, prompt_ids)
    text = tokenizer.decode(tail)
    ids: list[int] = []
    for label in labels:
        extended = tokenizer.encode(text + label, add_special_tokens=False)
        if extended[:-1] != tail or extended[-1] in ids:
            raise ValueError(
                f"label {label!r} is not one distinct token after this model's "
                "chat prompt"
            )
        ids.append(extended[-1])
    return tail, ids

select_read_strategy(model_config, supported_tasks)

Raise ValueError when the model cannot serve structured decisions.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def select_read_strategy(
    model_config: ModelConfig, supported_tasks: Sequence[SupportedTask]
) -> type[ReadStrategy]:
    """Raise ValueError when the model cannot serve structured decisions."""
    strategy = READ_STRATEGIES.get(model_config.architecture)
    if strategy is None:
        raise ValueError(
            "The structured decisions API does not support "
            f"{model_config.architecture}. Supported architectures: "
            f"{sorted(READ_STRATEGIES)}"
        )
    if strategy.task not in supported_tasks:
        raise ValueError(
            f"Structured decisions on {model_config.architecture} need the "
            f"{strategy.task} task"
        )
    if strategy.task == "generate" and model_config.logprobs_mode not in LOGPROBS_MODES:
        raise ValueError(
            "Structured decisions need --logprobs-mode raw_logprobs or "
            f"processed_logprobs, not {model_config.logprobs_mode}"
        )
    return strategy

type_label_ids(tokenizer, prompt_ids)

The reply's first token for each label of every registered question type. Raises ValueError as reply_label_ids does.

Source code in vllm/entrypoints/generate/structured_decisions/strategies.py

def type_label_ids(
    tokenizer: TokenizerLike, prompt_ids: Sequence[int]
) -> dict[str, int]:
    """The reply's first token for each label of every registered question
    type. Raises ValueError as reply_label_ids does."""
    label_ids: dict[str, int] = {}
    for qtype in QUESTION_TYPES.values():
        _, ids = reply_label_ids(tokenizer, prompt_ids, qtype.label_set)
        label_ids.update(zip(qtype.label_set, ids))
    return label_ids