Jev-Uslubidagi Bir Funksiyali O'ralish: Tushunish va Qullanish

Jev-Uslubidagi bir funksiyali o'ralish — bu yordamchi dasturlar va vision modellariga ham qo'llaniladigan kuchli vosita. Bu vosita yordamida LLM (Large Language Models) ning token ehtimolliklarini o'qish mumkin. Bu usul juda qadimiy, ammo ba'zi odamlar uchun yangi.

Asosiy Tamoyil

Asosiy tamoyil shundaki, foydalanuvchi biror savolni beradi va model bu savolga javob beradi. Misol uchun:

Holat: Buyurtmam parchalangan yetib keldi va men qaytarishni xohlardim. Savol: Qaysi jamo bu holatni hal qilish uchun javobgar? [A] hisob-kitob [B] yetkazib berish [C] qaytarish Faqat eng yaxshi variantning harfini javob bering.

Bu savolga javob berish uchun bir nechta JSON parametrlarini qo'shish kerak:

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 }

Bu parametrlar yordamida model harfni va boshqa tokenlarning ehtimolliklarini qaytaradi. Bu usul har bir savol uchun takrorlanadi.

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

Vision Modellariga Qo'llanilishi

Bu usul faqat matnli modellarga emas, vision modellarga ham qo'llanilishi mumkin. Jev-Uslubidagi so'rov formatida rasmlarni qo'shish mumkin. Misol uchun, webkamera ramkalarini yuborish va ularni baholash mumkin.

#!/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) == 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)

        # An omitted token cannot outrank the last returned alternative.
        # Allow zero only when their combined normalized probability is below 1e-6.
        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)}

        # Return the winning choice, probability of true, or expected ordinal level.
        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}


# Choose the server and model before opening the camera.
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()

# Point OpenCV's bundled Qt at the installed system fonts.
os.environ["QT_QPA_FONTDIR"] = "/usr/share/fonts/truetype/noto"

# Open the default Linux webcam with a small 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 continuously while a background worker scores one frame at a time.
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 a completed result, then submit the latest frame.
        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)
        # Measure throughput for evaluated frames, including 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()

Mening tajribamda, Gemma 4 12B modeli RTX 3090 grafik karta bilan taxminan 1 ramka sekundiga ishlaydi. Har bir ramka uchun uchta savol beriladi. OpenAI gpt-6-luna modeli esa taxminan 0.2 ramka sekundiga ishlaydi.

# 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

Python Misoli

Quyidagi Python misoli webkamera ramkalarini baholash uchun ishlatiladi:

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

#!/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 to

Xulosa

Jev-Uslubidagi bir funksiyali o'ralish — bu LLM va vision modellariga ham qo'llaniladigan kuchli vosita. Bu vosita yordamida modelning token ehtimolliklarini o'qish mumkin va bu usul juda qisqa vaqt ichida ishlaydi. Bu usulning eng yaxshi tomonlari orasida uning flexibilikligi va osonligi ajralib turadi.

Asl manba: allanrbo.blogspot.com