为LLMs提供类似Jev的单函数封装，包括视觉模型

# 类似 Jev 的 LLM 封装器（支持视觉模型） 来源：http://allanrbo.blogspot.com/2026/09/a-jev-like-wrapper-for-llms-including.html [](https://blogger.googleusercontent.com/img/b/R29vZ2xl/AVvXsEgslbgarcdEUUkIJWtO6Hzy1tHhrkARR5gWxsFPfsPk02zgLqPXsCoxkEIHa4ymkDB3aF-21KS9F0f4ScH9KVz-iN9GJ6yyTDC4rh8AaPG2QiCbZXDUNkRnRYLl8YDcuC8FsX9mkm-A9sb5s9V86UyC3A_I1Ai19idZYTgW1g8ZBLvwYirK97hZSw/s1369/webcam.jpeg) 我对 Jev (https://docs.typesafe.ai/introduction) 以及围绕它出现的可自托管项目（如 OpenJev (https://huggingface.co/openjev/openjev) 和 SemIf (https://github.com/TheoLeeCJ/SemIf-OpenJev)）很感兴趣。阅读相关资料让我了解到了一个巧妙的技巧：读取 LLM 的 token 概率。 对于某些人来说，这似乎是个老技巧了。例如 OpenAI 的 logprobs 实用手册 (https://developers.openai.com/cookbook/examples/using_logprobs)。但对我来说这是新知识。 我认为基本思路是编写一个这样的提示词： `` 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. `` 然后在兼容的聊天补全请求中添加几个 JSON 请求参数： `` { "max_completion_tokens": 1, "logprobs": true, "top_logprobs": 20 } `` LLM API 将返回字母以及模型为替代 token 提供的对数概率。 为每个问题重复此过程。强制只生成一个 token 避免了冗长的回答，并且速度非常快，尽管处理输入仍需要时间。不过，如果后端支持，每个问题共享的状态前缀可以被 KV 缓存。 有趣的是：这也适用于视觉模型。Jev 的文档化请求格式 (https://docs.typesafe.ai/introduction/quickstart) 目前只描述了文本/JSON 状态。在我的本地实验中，我添加了一个用于图像的 `attachments` 字段。 我的示例捕获网络摄像头帧，发送 base64 编码的 JPEG 图片，并打印一个表格：是否可见人物、我们在室内还是室外，以及场景的亮度如何？在我的 RTX 3090 上使用 Gemma 4 12B，每帧处理三个问题，我大约能达到 **1 帧/秒**。我也用 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": "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) # 附件是我们对 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(): # 将选项、布尔值和序数级别表示为字母选项。 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 需要 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 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) # 省略的 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 omitted non-negligible option scores for: {', '.join(missing)}") probabilities = {key: weight / total for key, weight in zip(options, weights)} # 返回获胜的选择、为真的概率或期望的序数级别。 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("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) # 持续预览，同时后台工作线程一次处理一帧。 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 # 打印完成的结果，然后提交最新的帧。 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("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() `` 该脚本处理 API 差异：llama\.cpp 使用聊天补全，而 OpenAI 使用响应来获取替代选项。 我通过 llama\.cpp 运行了 Gemma 4 12B QAT。在安装了 NVIDIA 驱动程序、`curl`、`zstd` 和 `uv` 的 Linux 上： `` # 模型（约 7 GB）和多模态投影器（约 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 # 适用于 RTX 3090（CUDA 架构 86）的独立 llama.cpp 二进制文件。 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 # 或使用 OpenAI，需在环境中设置 OPENAI_API_KEY。 uv run webcam.py https://api.openai.com/v1 gpt-6-luna ``