mixle.task.vlm module

A tiny provider-agnostic VLM surface, wired directly into mixle.enumeration’s descending-probability search.

Sibling of mixle.task.llm, extended with an image. A VLM is anything with next_logprobs(image, prefix) -> [(token, log_prob), ...] – the SAME shape mixle.enumeration.best_first_decode() / mixle.enumeration.quantized_best_first_decode() already expect from any autoregressive scorer, so binding an image into that shape (OpenAICompatVLM.next_logprobs_for()) is all “VLM enumeration support” needs to be: nothing about best_first_decode itself is vision-specific.

Scope, deliberately: this targets an open-weight vision-language model served behind an OpenAI-compatible /v1/chat/completions endpoint that returns real per-token logprobs (vLLM, TGI, and similar self-hosted stacks serving e.g. LLaVA / Qwen-VL / …). Proprietary hosted vision APIs (GPT-4V, Claude vision, Gemini vision) generally do not expose true per-token logprobs for image-conditioned generation, so mixle.enumeration’s descending-probability guarantee only holds against a genuine logprob-serving endpoint – OpenAICompatVLM does not attempt to approximate that guarantee against a black-box API that cannot honor it.

Two things this file gives you, both built on the one real network primitive (OpenAICompatVLM.next_logprobs()):

  • Free-form top-k decoding, exact and lazy, via the existing engine:

    vlm = OpenAICompatVLM("http://localhost:8000/v1", "llava-onevision")
decode = vlm.next_logprobs_for(image, prompt="Describe this image in one sentence.")
for tokens, log_prob in best_first_decode(decode, eos="<|eot_id|>", max_len=40, max_results=5):
    print("".join(tokens), log_prob)   # the 5 highest-probability captions, best first

  • Ranking a fixed candidate set by the model’s own teacher-forced probability, via mixle.enumeration.top_k_scored():

    score = score_fn_for(decode)
top_k_scored([("cat",), ("dog",), ("bird",)], score, k=3)

Teacher-forced candidate scoring costs one next_logprobs call per token (no batched echo/teacher-forcing primitive is assumed to exist on the server) – this is stated up front, not hidden behind a fast-looking API. Candidates must be pre-tokenized (Sequence[str] of the SAME token pieces the server’s own tokenizer would produce): a naive whitespace/character split would silently misalign with a real BPE tokenizer and score the wrong thing, so no such convenience split is provided here.

class VLM(*args, **kwargs)[source]

Bases: Protocol

Anything that can score an image-conditioned next-token continuation.

class CallableVLM(fn)[source]

Bases: object

Wrap a plain fn(image, prefix) -> [(token, log_prob), ...] as a VLM – local models and tests.

Parameters:

fn (Callable[[Any, tuple[str, ...]], Iterable[tuple[str, float]]])

next_logprobs_for(image)[source]

Bind image into the next_logprobs(prefix) shape mixle.enumeration expects directly.

Parameters:

image (Any)

Return type:

Callable[[tuple[str, …]], Iterable[tuple[str, float]]]

class OpenAICompatVLM(base_url, model, *, api_key=None, top_logprobs=20, timeout=60.0, continue_key='continue_final_message', continue_value=True, extra_body=None)[source]

Bases: object

A VLM backed by an OpenAI-compatible /v1/chat/completions endpoint that returns real per-token logprobs for an open-weight vision-language model (a vLLM- or TGI-served LLaVA / Qwen-VL / … deployment). See the module docstring for why this deliberately does not target proprietary hosted vision APIs.

Continuing a partial completion (every next_logprobs call after the first token of a decode) needs the server to prefill the given prefix rather than start generation fresh; this uses vLLM’s continue_final_message extension by default (append the prefix as a partial assistant message, set that flag). Pass continue_key/continue_value to target a server with a different convention.

Parameters:

  • base_url (str)

  • model (str)

  • api_key (str | None)

  • top_logprobs (int)

  • timeout (float)

  • continue_key (str)

  • continue_value (Any)

  • extra_body (dict[str, Any] | None)

next_logprobs(image, prefix, *, prompt, system=None)[source]

One image-conditioned next-token distribution given the tokens generated so far (prefix).

Parameters:
Return type:

list[tuple[str, float]]

next_logprobs_for(image, prompt, *, system=None)[source]

Bind image/prompt into the next_logprobs(prefix) -> [(token, log_prob), ...] shape mixle.enumeration.best_first_decode() / mixle.enumeration.quantized_best_first_decode() expect directly – the whole bridge from “an image and a question” to “enumerate the top-k answers”.

Parameters:

  • image (Any)

  • prompt (str)

  • system (str | None)

Return type:

Callable[[tuple[str, …]], Iterable[tuple[str, float]]]

score_candidate(next_logprobs_fn, candidate_tokens)[source]

Teacher-forced total log-probability of candidate_tokens under next_logprobs_fn.

Walks one token at a time, reading off the ACTUAL log-probability of the candidate’s own next token at each step – never approximated or guessed. If a step’s returned continuations do not include the candidate’s token (e.g. it fell outside top_logprobs), returns -inf rather than silently dropping or padding the score with a made-up value: that is a real “this candidate wasn’t even considered by the model at that step” outcome, not a bug to hide.

Parameters:
Return type:

float

score_fn_for(next_logprobs_fn)[source]

Bind a next_logprobs function into the score(candidate) -> float shape mixle.enumeration.top_k_scored() expects directly, for ranking a fixed candidate set.

Parameters:

next_logprobs_fn (Callable[[tuple[str, ...]], Iterable[tuple[str, float]]])

Return type:

Callable[[Sequence[str]], float]