LLM 与视觉模型的 Jev 风格单函数封装器

我对 Jev 以及围绕它出现的可自托管项目很感兴趣，比如 OpenJev 和 SemIf。在读这些资料时，我了解到一个很巧妙的技巧：读取 LLM 的 token 概率。

显然对一些人来说这是老把戏了，比如可以看看 OpenAI 的 logprobs cookbook。但对我来说是新鲜事。

基本思路是写一个这样的 prompt：

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.

然后在兼容的 Chat Completions 请求中加上几个 JSON 参数：

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

LLM API 会返回那个字母，以及模型对各个备选 token 的对数概率。

对每个问题重复这一过程。强制只生成一个 token，既避免了冗长的回答，速度也极快——不过处理输入仍需要时间。如果后端支持，多个问题共享的 state 前缀还可以利用 KV cache。

有意思的是：这对视觉模型同样有效。Jev 的文档中的请求格式目前只描述了文本/JSON 形式的 state，我在本地实验时自己加了一个用于图片的 attachments 字段。

我的示例会捕获摄像头画面，发送 base64 编码的 JPEG，然后打印一张表：画面中是否有人、是室内还是室外、光线亮度如何？在我的 RTX 3090 上跑 Gemma 4 12B，每帧问三个问题，大约能达到 1 FPS。用 OpenAI 的 gpt-6-luna 跑则只有约 0.2 FPS，大概是因为我没做任何优化，每个问题的每帧请求都单独走了一遍他们的连接。

专用视觉模型固然效率更高，但我更欣赏的是这里的灵活性：只需通过纯文本描述即可更改条件。

这是一个独立的 Python 示例（OpenCV 仅用于便捷地访问摄像头，并不涉及实际的视觉处理）：

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["opencv-python"]
# ///
"""使用 llama.cpp 或 OpenAI 预览并评分摄像头画面。

uv run webcam.py
uv run webcam.py https://api.openai.com/v1 gpt-6-luna
OpenAI 通过环境变量 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 是我们在 Jev 请求格式中的自定义扩展。
data = json.loads("""
{
    "state": "检查此摄像头画面。仅判断画面中可见的内容。",
    "attachments": [],
    "questions": {
        "person": {
            "type": "noul",
            "instructions": "是否有人物可见？"
        },
        "plant": {
            "type": "noul",
            "instructions": "是否有植物可见？"
        },
        "setting": {
            "type": "choice",
            "instructions": "摄像头位于何处？",
            "criteria": {
                "indoors": null,
                "outdoors": null,
                "unclear": null
            }
        },
        "light": {
            "type": "score",
            "instructions": "场景有多亮？",
            "criteria": [
                "dark",
                "dim",
                "bright"
            ]
        }
    }
}
""")

def score(data, url, model):
    state = data["state"]
    if not isinstance(state, str):
        state = json.dumps(state)

    # attachments 是我们对 Jev 风格请求格式的扩展：
    # 图像文件路径或 base64 数据 URL。为所有问题一次性加载。
    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"不支持的图像文件：{path}")
        encoded = base64.b64encode(path.read_bytes()).decode()
        images.append(f"data:{mime_type};base64,{encoded}")

    # API 密钥仅发送给 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():
        # 将选项、布尔值和有序等级表示为字母选项。
        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"未知的问题类型：{question['type']}")
        if not 2 <= len(options) <= 20:
            raise ValueError("每个问题需提供 2 到 20 个标准项。")
        letters = "ABCDEFGHIJKLMNOPQRST"[:len(options)]

        # 请求单个选项字母，使其 logprob 代表该选项。
        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\n仅回答最佳选项的字母。"

        # OpenAI 需要 Responses 接口以获取足够的备选答案；llama.cpp 需要 Chat 接口以获取 logprobs。
        # top_p=1 避免剪枝备选答案。
        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,
            }

        # 发送请求并读取第一个输出 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}

        # 归一化返回的选项分数；缺失的选项初始值为零。
        missing = [letter for letter in letters if letter not in logprobs or logprobs[letter] <= -9999]
        if len(missing) == len(letters):
            raise ValueError("API 未返回任何选项的有效分数")
        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)

        # 被省略的 token 不能超过最后返回的备选答案的排名。
        # 仅当其归一化概率之和低于 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 省略了非可忽略的选项分数：{', '.join(missing)}")
        probabilities = {key: weight / total for key, weight in zip(options, weights)}

        # 返回获胜选项、true 的概率，或期望的有序等级。
        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}

# 在打开摄像头前选择服务器和模型。
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()

# 将 OpenCV 自带的 Qt 指向已安装的系统字体。
os.environ["QT_QPA_FONTDIR"] = "/usr/share/fonts/truetype/noto"

# 使用较小的捕获缓冲区打开默认 Linux 摄像头。
camera = cv2.VideoCapture(0, cv2.CAP_V4L2)
if not camera.isOpened():
    raise RuntimeError("无法打开 /dev/video0")
camera.set(cv2.CAP_PROP_BUFFERSIZE, 1)
print(f"Webcam -> {args.model}. Noul: 是 %; score: 值/最大值. 按 Ctrl-C 或 Esc 停止。", flush=True)
print(f"{'time':<8}" + "".join(f"{name:>10}" for name in data["questions"]) + f"{'fps':>10}", flush=True)

# 持续预览，同时后台工作线程一次评分一帧画面。
executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
pending = None
try:
    while True:
        ok, frame = camera.read()
        if not ok:
            raise RuntimeError("无法读取摄像头画面")
        cv2.imshow("Webcam", frame)
        if cv2.waitKey(1) == 27 or cv2.getWindowProperty("Webcam", cv2.WND_PROP_VISIBLE) < 1:
            break

        # 打印已完成的结果，然后提交最新帧。
        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)
        # 测量已评估帧的吞吐量，包括图像编码。
        started = time.perf_counter()
        captured = datetime.datetime.now().strftime("%H:%M:%S")
        ok, jpeg = cv2.imencode(".jpg", frame)
        if not ok:
            raise RuntimeError("无法编码摄像头画面")
        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("\n已停止。")
finally:
    camera.release()
    cv2.destroyAllWindows()
    executor.shutdown()

这个脚本处理了 API 上的差异：llama.cpp 用的是 Chat Completions 接口，而 OpenAI 用的是 Responses 接口，这样才能让它展示候选结果。

我用 llama.cpp 跑了 Gemma 4 12B QAT。在安装了 NVIDIA 驱动、curl、zstd 和 uv 的 Linux 上：

# 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 示例保存为 webcam.py。然后另开一个终端，在该目录下运行：

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