Een Jev-achtige wrapper voor LLM's, inclusief vision-modellen
Voor sommige mensen is dit een oude truc (zie bijvoorbeeld de logprobs cookbook van OpenAI), maar voor mij was het nieuw. De basisidee is om een prompt als volgt op te stellen:
State: Mijn bestelling is beschadigd aangekomen en ik wil mijn geld terug. Question: Welk team moet dit afhandelen? [A] billing [B] shipping [C] returns Antwoord uitsluitend met de letter van de beste optie.
Vervolgens voeg je een paar JSON-aanvraagparameters toe aan een compatibel Chat Completions-verzoek:
{
"max_completion_tokens": 1,
"logprobs": true,
"top_logprobs": 20
}De LLM API geeft vervolgens de letter terug, samen met de log-probabiliteiten van het model voor alternatieve tokens.
Dit proces wordt herhaald voor elke vraag. Door het model te dwingen slechts één token te genereren, wordt een langdradig antwoord voorkomen en is het proces zeer snel, hoewel het verwerken van de input nog steeds tijd kost. Indien de backend dit ondersteunt, kan een gedeelde state-prefix voor elke vraag in de KV-cache worden bewaard.
Toepassing op vision-modellen
Het interessante is dat dit ook werkt met vision-modellen. Het gedocumenteerde aanvraagformaat van Jev beschrijft momenteel alleen tekst/JSON-state. Voor mijn lokale experimenten heb ik een attachments-veld voor afbeeldingen toegevoegd.
Mijn voorbeeld legt webcamframes vast, verzendt base64 JPEGs en print een tabel met de volgende vragen: is er een persoon zichtbaar, bevinden we ons binnen of buiten, en hoe helder is de scène?
Met Gemma 4 12B op mijn RTX 3090 haal ik ongeveer 1 frame per seconde (FPS), met drie vragen per frame. Ik heb het ook getest met OpenAI gpt-6-luna en haalde ongeveer 0,2 FPS. Dit is vermoedelijk omdat ik geen moeite heb gedaan om de kosten van een aparte verbinding via hun systeem per vraag per frame te vermijden.
Hoewel gespecialiseerde computer vision-modellen zeker veel efficiënter zijn, is het voordeel hier de flexibiliteit: je kunt een conditie wijzigen door deze simpelweg in platte tekst te beschrijven.
Python-voorbeeld
Hieronder staat het standalone Python-voorbeeld. OpenCV wordt hier uitsluitend gebruikt voor gemakkelijke toegang tot de webcam, niet voor daadwerkelijke computer vision.
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["opencv-python"]
# ///
"""Preview en score webcam frames met llama.cpp of OpenAI.
uv run webcam.py
uv run webcam.py https://api.openai.com/v1 gpt-6-luna
OpenAI leest OPENAI_API_KEY.
"""
import argparse
import base64
import concurrent.futures
import datetime
import json
import math
import mimetypes
import os
import pathlib
import time
import urllib.parse
import urllib.request
import cv2
# 'attachments' is onze eigen toevoeging aan het Jev-aanvraagformaat.
data = json.loads("""
{
"state": "Inspect this webcam frame. Judge only what is visibly present.",
"attachments": [],
"questions": {
"person": {
"type": "noul",
"instructions": "Is a person visible?"
},
"plant": {
"type": "noul",
"instructions": "Is a plant visible?"
},
"setting": {
"type": "choice",
"instructions": "Where is the camera?",
"criteria": {
"indoors": null,
"outdoors": null,
"unclear": null
}
},
"light": {
"type": "score",
"instructions": "How bright is the scene?",
"criteria": [
"dark",
"dim",
"bright"
]
}
}
}
""")
def score(data, url, model):
state = data["state"]
if not isinstance(state, str):
state = json.dumps(state)
# Attachments zijn onze uitbreiding op het Jev-stijl aanvraagformaat:
# afbeeldingpaden of base64 data URLs. Eén keer laden voor alle vragen.
images = []
for attachment in data.get("attachments", []):
if attachment.startswith("data:image/"):
images.append(attachment)
continue
path = pathlib.Path(attachment).expanduser()
mime_type, _ = mimetypes.guess_type(path)
if mime_type not in {"image/png", "image/jpeg", "image/webp", "image/gif"}:
raise ValueError(f"Unsupported image file: {path}")
encoded = base64.b64encode(path.read_bytes()).decode()
images.append(f"data:{mime_type};base64,{encoded}")
# Verstuur de API-sleutel alleen naar OpenAI.
is_openai = urllib.parse.urlsplit(url).hostname == "api.openai.com"
headers = {"Content-Type": "application/json"}
if is_openai:
headers["Authorization"] = "Bearer " + os.environ["OPENAI_API_KEY"]
answers = {}
for name, question in data["questions"].items():
# Representeer keuzes, booleans en ordinale niveaus als gelabelde opties.
if question["type"] == "choice":
options = question["criteria"]
elif question["type"] == "noul":
options = {"true": None, "false": None} | question.get("criteria", {})
elif question["type"] == "score":
options = {str(i): description for i, description in enumerate(question["criteria"])}
else:
raise ValueError(f"Unknown question type: {question['type']}")
if not 2 <= len(options) <= 20:
raise ValueError("Provide 2 to 20 criteria per question.")
letters = "ABCDEFGHIJKLMNOPQRST"[:len(options)]
# Vraag om één optie-letter, zodat de logprob die optie representeert.
instructions = question["instructions"]
if not isinstance(instructions, str):
instructions = json.dumps(instructions)
lines = [f"State:\n{state}\n\nQuestion: {instructions}\nOptions:"]
for letter, (key, description) in zip(letters, options.items()):
line = f"[{letter}] {key}"
if description is not None:
line += f": {description}"
lines.append(line)
prompt = "\n".join(lines) + "\n\nAnswer with the letter of the best option only."
# OpenAI heeft 'Responses' nodig voor voldoende alternatieven;
# llama.cpp heeft 'Chat' nodig voor logprobs.
# top_p=1 voorkomt het snoeien van alternatieven.
if is_openai:
endpoint = "/responses"
content = [{"type": "input_text", "text": prompt}]
content.extend({"type": "input_image", "image_url": image} for image in images)
body = {
"model": model,
"input": [{"role": "user", "content": content}],
"reasoning": {"effort": "none"},
"max_output_tokens": 16,
"top_p": 1,
"top_logprobs": 20,
"include": ["message.output_text.logprobs"],
}
else:
endpoint = "/chat/completions"
content = [{"type": "text", "text": prompt}]
content.extend({"type": "image_url", "image_url": {"url": image}} for image in images)
body = {
"model": model,
"messages": [{"role": "user", "content": content}],
"max_completion_tokens": 1,
"temperature": 0,
"reasoning_effort": "none",
"logprobs": True,
"top_logprobs": 1024,
}
# Verstuur het verzoek en lees de alternatieven van het eerste output-token.
request = urllib.request.Request(
url.rstrip("/") + endpoint,
headers=headers,
data=json.dumps(body).encode(),
)
with urllib.request.urlopen(request) as response:
result = json.load(response)
if is_openai:
message = next(item for item in result["output"] if item["type"] == "message")
candidates = message["content"][0]["logprobs"][0]["top_logprobs"]
else:
candidates = result["choices"][0]["logprobs"]["content"][0]["top_logprobs"]
logprobs = {item["token"]: item["logprob"] for item in candidates}
# Normaliseer de teruggegeven optie-scores; ontbrekende opties krijgen aanvankelijk nul.
missing = [letter for letter in letters if letter not in logprobs or logprobs[letter] <= -9999]
if len(missing) == len(letters):
raise ValueError("API did not return usable scores for any option")
peak = max(logprobs[letter] for letter in letters if letter not in missing)
weights = [math.exp(logprobs[letter] - peak) if letter not in missing else 0 for letter in letters]
total = sum(weights)
# Een weggelaten token kan niet hoger scoren dan het laatste teruggegeven alternatief.
if missing:
cutoff = min(value for value in logprobs.values() if value > -9999)
missing_weight = len(missing) * math.exp(cutoff - peak)
if missing_weight / (total + missing_weight) >= 1e-6:
raise ValueError(f"API omitted non-negligible option scores for: {', '.join(missing)}")
probabilities = {key: weight / total for key, weight in zip(options, weights)}
# Geef de winnende keuze, de kans op 'true', of het verwachte ordinale niveau terug.
if question["type"] == "choice":
answers[name] = {
"type": "choice",
"choice": max(probabilities, key=probabilities.get),
"probabilities": probabilities,
}
elif question["type"] == "noul":
answers[name] = {"type": "noul", "noul": probabilities["true"]}
else:
answers[name] = {
"type": "score",
"score": sum(int(key) * probability for key, probability in probabilities.items()),
"legend": options,
"probabilities": probabilities,
}
return {"answers": answers}
# Kies de server en het model voordat de camera wordt geopend.
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("url", nargs="?", default="http://localhost:8060/v1")
parser.add_argument("model", nargs="?", default="gemma-4-12b")
args = parser.parse_args()
# Wijs de gebundelde Qt van OpenCV toe aan de geïnstalleerde systeemlettertypen.
os.environ["QT_QPA_FONTDIR"] = "/usr/share/fonts/truetype/noto"
# Open de standaard Linux webcam met een kleine capture buffer.
camera = cv2.VideoCapture(0, cv2.CAP_V4L2)
if not camera.isOpened():
raise RuntimeError("Could not open /dev/video0")
camera.set(cv2.CAP_PROP_BUFFERSIZE, 1)
print(f"Webcam -> {args.model}. Noul: yes %; score: value/max. Ctrl-C or Esc to stop.", flush=True)
print(f"{'time':<8}" + "".join(f"{name:>10}" for name in data["questions"]) + f"{'fps':>10}", flush=True)
# Preview continu terwijl een background worker één frame per keer scoort.
executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
pending = None
try:
while True:
ok, frame = camera.read()
if not ok:
raise RuntimeError("Could not read a webcam frame")
cv2.imshow("Webcam", frame)
if cv2.waitKey(1) == 27 or cv2.getWindowProperty("Webcam", cv2.WND_PROP_VISIBLE) < 1:
break
# Print een voltooid resultaat, en dien vervolgens het nieuwste frame in.
if pending is not None:
if not pending.done():
continue
result = pending.result()
columns = []
for name in data["questions"]:
answer = result["answers"][name]
if answer["type"] == "noul":
value = f"{answer['noul']:.1%}"
elif answer["type"] == "choice":
value = answer["choice"]
else:
value = f"{answer['score']:.2f}/{len(data['questions'][name]['criteria']) - 1}"
columns.append(f"{value:>10}")
columns.append(f"{1 / (time.perf_counter() - started):>10.2f}")
print(captured + "".join(columns), flush=True)
# Meet de throughput voor geëvalueerde frames, inclusief image encoding.
started = time.perf_counter()
captured = datetime.datetime.now().strftime("%H:%M:%S")
ok, jpeg = cv2.imencode(".jpg", frame)
if not ok:
raise RuntimeError("Could not encode the webcam frame")
image = "data:image/jpeg;base64," + base64.b64encode(jpeg.tobytes()).decode()
data["attachments"] = [image]
pending = executor.submit(score, data, args.url, args.model)
except KeyboardInterrupt:
print("\nStopped.")
finally:
camera.release()
cv2.destroyAllWindows()
executor.shutdown()Het script handelt de API-verschillen af: llama.cpp gebruikt Chat Completions en OpenAI gebruikt Responses om de alternatieven te tonen.
Installatie en uitvoering
Ik heb Gemma 4 12B QAT gedraaid via llama.cpp. Op Linux met NVIDIA-drivers, curl, zstd en uv geïnstalleerd, gebruik je de volgende commando's:
# Model (~7 GB) en multimodal projector (~175 MB)
mkdir -p ~/models/gemma-4-12b/
cd ~/models/gemma-4-12b/
curl -fL -C - -o gemma-4-12b-it-qat-q4_0.gguf https://huggingface.co/google/gemma-4-12B-it-qat-q4_0-gguf/resolve/main/gemma-4-12b-it-qat-q4_0.gguf
curl -fL -C - -o mmproj-gemma-4-12b-it-qat-q4_0.gguf https://huggingface.co/google/gemma-4-12B-it-qat-q4_0-gguf/resolve/main/mmproj-gemma-4-12b-it-qat-q4_0.gguf
# Standalone llama.cpp binary voor RTX 3090 (CUDA architecture 86)
curl -fL -o llama.zst https://huggingface.co/buckets/ggml-org/install.sh/resolve/b11160/x86_64/linux/cuda/86/llama-app.zst
mkdir -p ~/bin/
zstd -d llama.zst -o ~/bin/llama
chmod +x ~/bin/llama
~/bin/llama serve --models-dir ~/models/ --port 8060Sla het Python-voorbeeld op als webcam.py. Voer in een andere terminal vanuit die directory het volgende uit:
uv run webcam.py http://localhost:8060/v1 gemma-4-12b
# Of gebruik OpenAI, met OPENAI_API_KEY ingesteld in je omgeving.
uv run webcam.py https://api.openai.com/v1 gpt-6-luna
Groetjes,