Hàm đơn giản hóa giúp đọc xác suất lựa chọn của LLM và mô hình thị giác

State: My order arrived broken and I want a refund.
Question: Which team should handle this?
[A] billing
[B] shipping
[C] returns
Answer with the letter of the best option only.

{
  "max_completion_tokens": 1,
  "logprobs": true,
  "top_logprobs": 20
}

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["opencv-python"]
# ///
"""Preview and score webcam frames with llama.cpp or OpenAI.

uv run webcam.py
uv run webcam.py https://api.openai.com/v1 gpt-6-luna
OpenAI reads 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 our custom addition to the Jev request format.
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 are our extension to the Jev-style request format:
    # image file paths or base64 data URLs. Load them once for all questions.
    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}")

    # Send the API key only to 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():
        # Represent choices, booleans, and ordinal levels as lettered options.
        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)]

        # Ask for a single option letter, so its logprob represents that option.
        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 needs Responses for enough alternatives; llama.cpp needs Chat for logprobs.
        # top_p=1 avoids pruning alternatives.
        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,
            }

        # Send the request and read the first output token's alternatives.
        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}

        # Normalize the returned option scores; missing options initially get zero.
        missing = [letter for letter in letters if letter not in logprobs or logprobs[letter] <= -9999]
        if len(missing) == l

# Model (~7 GB) and 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 for 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 8060

uv run webcam.py http://localhost:8060/v1 gemma-4-12b
# Or use OpenAI, with OPENAI_API_KEY set in your environment.
uv run webcam.py https://api.openai.com/v1 gpt-6-luna