AI・機械学習
ビジョンモデルを含むLLMの単一関数Jevライクラッパー
A single function Jev-like wrapper for LLMs, including vision models (allanrbo.blogspot.com)
要約
この記事では、LLM(大規模言語モデル)の出力を分析するための新しいアプローチとして、Jevライクなラッパーを紹介しています。このラッパーは、プロンプトに特定の形式で質問を埋め込み、モデルのトークン確率(logprobs)を取得することで、単一トークンでの回答を強制し、分析を高速化します。さらに、この手法は画像入力に対応するビジョンモデルにも拡張可能であり、Webカメラの映像をリアルタイムで分析し、人物の有無、屋内/屋外、明るさなどを判断するデモンストレーションコードも提供されています。
全文翻訳
メインへスキップ | サイドバーへスキップ
Allan's Blog
Allan Riordan Boll's blog
金曜日, 2026年9月25日
ビジョンモデルを含むLLMの単一関数Jevライクラッパー
Jevと、OpenJevやSemIfのようなその周辺で出現しているセルフホスト可能なプロジェクトに興味をそそられました。それらについて読むことで、LLMのトークン確率を読むという、ある人にとっては古いトリックですが、私にとっては新しいトリックに出会いました。例えば、OpenAIのlogprobsクックブックを参照してください。基本的な考え方は、次のようなプロンプトを書くことだと思います。
状態: 私の注文品が破損して届きました。返金を希望します。質問: どのチームがこれを担当すべきですか? [A] 請求 [B] 配送 [C] 返品 最良の選択肢の文字のみで回答してください。
そして、互換性のあるChat CompletionsリクエストにいくつかのJSONリクエストパラメータを追加します。
{ "max_completion_tokens": 1, "logprobs": true, "top_logprobs": 20 }
LLM APIは、その文字と、代替トークンに対するモデルの対数確率を返します。各質問で繰り返します。1つのトークンのみを生成するように強制することで、長い回答を避け、非常に高速になりますが、入力の処理には依然として時間がかかります。ただし、各質問に対して、バックエンドがサポートしていれば、共有状態のプレフィックスをKVキャッシュできます。
面白い部分: これはビジョンモデルでも機能します。
Jevのドキュメント化されたリクエスト形式は、現在、テキスト/JSON状態のみを記述しています。私のローカル実験のために、画像用のattachmentsフィールドを追加しました。私の例では、Webカメラのフレームをキャプチャし、base64エンコードされたJPEGを送信し、テーブルを表示します。人物が見えるか、屋内か屋外か、シーンはどれくらい明るいか? Gemma 4 12BをRTX 3090で実行すると、1秒あたり約1フレーム、1フレームあたり3つの質問が得られます。OpenAI gpt-6-lunaでも実行しましたが、約0.2 FPSでした。おそらく、質問ごとにフレームごとにシステムを介した個別の接続のコストを回避する努力をしなかったためでしょう。専用のコンピュータビジョンモデルは間違いなくはるかに効率的ですが、私がここで気に入っているのは柔軟性です。条件をプレーンテキストで記述することで変更できます。
以下はスタンドアロンのPython例です(OpenCVはWebカメラへの便利なアクセスに使用されるだけで、実際のコンピュータビジョンには使用されません)。
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["opencv-python"]
# ///
"""llama.cppまたはOpenAIでWebカメラフレームをプレビューおよびスコアリングします。 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": "このWebカメラフレームを検査してください。目に見えるものだけを判断してください。",
"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"Unsupported image file: {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():
# choices, booleans, and ordinal levels を文字オプションとして表現します。
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)]
# 単一オプションの文字を要求するため、その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\nAnswer with the letter of the best option only."
# OpenAIは十分な代替手段のためのResponsesを必要とします;llama.cppはlogprobsのためのChatを必要とします。
# 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,
}
# リクエストを送信し、最初の出力トークンの代替手段を読み取ります。
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 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)
# 省略されたトークンは、最後に返された代替手段を上回ることはできません。
# それらの合計正規化確率が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)}
# 勝った選択肢、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] = {