formula_img2latex/formula_img2latex/engines/vlm.py
dschlueter 388024a17b feat: 0.2.0 – Konsens Qwen = Gemma = GLM als Standard, Gehrke komplett verarbeitet
- Dritte Lesart (GLM-4.6V über OpenRouter, ohne Nachdenken) muss zustimmen;
  keine 2:1-Annahme; fehlender Schlusspunkt der dritten Lesart zählt nicht
- run in Paketen zu 50 mit Statuszeile, beide lokalen Modelle parallel,
  GLM nur für Einige; nur die übernommene Lesart wird gegen das Bild geprüft
- Bild-Veto: Ähnlichkeit >= 0,65, Schriftgrößen-Prüfung abgeschaltet
- Antworten auf 800 Tokens begrenzt; Prompt-Version und Nachdenk-Einstellung
  im Cache-Schlüssel; Verfahrensversion für Prüfergebnisse
- Behoben: Gleichungsnummern aus alten Lesarten ("(3.7)"), gleichzeitiges
  Speichern zweier Engines, Absturz bei Formeln ohne aktuelle Lesart
- apply: deutsches Dezimalkomma 3{,}6; Review-Galerie zeigt aktuelle Lesarten
- tools/vergleich.py auf Konsens-Ablauf umgestellt
- Doku: README, BEDIENUNGSANLEITUNG, neues CHANGELOG

Gehrke-Brückenkurs: 942 von 1505 Formelbildern automatisch ersetzt (63 %),
62 von 62 von Hand geprüften korrekt; Pandoc/MathML 945 von 945.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-07 07:58:16 +02:00

250 lines
11 KiB
Python

"""Vision-language model over an OpenAI-compatible chat-completions endpoint."""
from __future__ import annotations
import base64
import io
import json
import re
import time
from dataclasses import dataclass
import numpy as np
import requests
from PIL import Image
from ..errors import EXIT_TOOL, FormulaError
from ..imageio import gray_to_ink, load_gray
MIN_HEIGHT = 96 # small formula images are enlarged to at least this height for the model
MAX_UPSCALE = 8
TIMEOUT_S = 120
MAX_TOKENS = 800 # long multi-line formulas need ~400; runaway repetitions are cut off (they are useless anyway)
PROMPT_VERSION = 2 # bump when PROMPT changes: cached readings of older prompts are asked again
PROMPT = """Transcribe the mathematical formula in the image into LaTeX (amsmath, amssymb).
Rules:
- Reproduce exactly what is printed: every symbol, digit, sign, sub- and superscript, prime, accent and bracket.
- Do not simplify, complete, reorder or correct anything. Do not add a trailing period unless one is printed.
- Check every letter and digit for BOLD type. Bold symbols (vectors, matrices, the zero vector) are common
and easy to miss: write them as \\boldsymbol{v}, \\boldsymbol{0}. Non-bold symbols stay plain.
- Use \\mathbb, \\mathcal, \\mathrm etc. where the print uses such fonts.
- Use \\frac for fractions, \\sqrt for roots, \\begin{pmatrix}/\\begin{array}/\\begin{cases} for matrices and cases,
\\begin{aligned} for several aligned lines.
- Write an equation number such as (3.7) at the right margin as \\tag{3.7}.
- Do not wrap the result in $ or \\[ \\].
- If the image is not a formula (photo, diagram, plot, plain text), set is_formula to false and latex to "".
- confidence: how sure you are (0-100) that every symbol of your LaTeX matches the image.
{context}
Answer with JSON only: {"is_formula": true, "latex": "...", "confidence": 0}"""
CONTEXT_LINE = "- The formula is {where} this text from the book (for orientation only, do not copy it): \"{text}\"\n"
@dataclass
class Recognition:
engine: str
model: str
tex: str | None
is_formula: bool | None
raw: str = ""
error: str | None = None
seconds: float = 0.0
tokens_in: int = 0
tokens_out: int = 0
p_min: float | None = None # smallest token probability inside the LaTeX string
p_mean: float | None = None # geometric mean of those token probabilities
self_conf: float | None = None # the model's own 0-100 estimate (stored, not trusted)
def to_json(self) -> dict:
return {k: v for k, v in self.__dict__.items()}
def prepare_image(data: bytes, name: str = "") -> bytes:
"""Flatten transparency onto white, enlarge tiny formulas, add a margin; PNG bytes."""
ink = gray_to_ink(load_gray(data, name))
gray = (255 - np.clip(ink * 255, 0, 255)).astype(np.uint8)
im = Image.fromarray(gray)
if im.height < MIN_HEIGHT:
f = min(MAX_UPSCALE, max(1, int(np.ceil(MIN_HEIGHT / max(im.height, 1)))))
im = im.resize((im.width * f, im.height * f), Image.LANCZOS)
pad = max(8, im.height // 6)
canvas = Image.new("L", (im.width + 2 * pad, im.height + 2 * pad), 255)
canvas.paste(im, (pad, pad))
buf = io.BytesIO()
canvas.save(buf, "PNG")
return buf.getvalue()
def parse_answer(text: str) -> tuple[str | None, bool | None]:
"""Extract (latex, is_formula) from a model answer (JSON, fenced JSON, or bare LaTeX)."""
s = text.strip()
s = re.sub(r"^```(?:json|latex|tex)?\s*|\s*```$", "", s, flags=re.S).strip()
m = re.search(r"\{.*\}", s, re.S)
if m:
blob = m.group(0)
# Models often write LaTeX with single backslashes inside JSON: "\frac" would become a
# form feed + "rac", "\nu" a newline + "u". Such control characters mean: re-read the
# blob with every lone backslash taken literally.
literal = re.sub(r'\\\\|\\"|\\', lambda mm: "\\\\" if mm.group(0) == "\\" else mm.group(0), blob)
for attempt in (blob, literal):
try:
data = json.loads(attempt)
except ValueError:
continue
if not isinstance(data, dict):
continue
tex = data.get("latex")
if attempt is blob and isinstance(tex, str) and re.search(r"[\x00-\x1f]", tex):
continue
isf = data.get("is_formula")
return (tex.strip() if isinstance(tex, str) and tex.strip() else None,
bool(isf) if isf is not None else None)
mm = re.search(r'"latex"\s*:\s*"(.*)"\s*\}?\s*$', blob, re.S)
if mm:
return mm.group(1).replace('\\"', '"').replace("\\\\", "\\").strip() or None, None
if s and "{" not in s[:1]:
return s, None
return None, None
def parse_confidence(text: str) -> float | None:
m = re.search(r'"confidence"\s*:\s*"?(\d+(?:\.\d+)?)', text)
if not m:
return None
v = float(m.group(1))
return v * 100 if v <= 1.0 and "." in m.group(1) else v
def _latex_span(content: str) -> tuple[int, int] | None:
"""Character span of the JSON string value of "latex" in the model answer."""
m = re.search(r'"latex"\s*:\s*"', content)
if not m:
return None
i = m.end()
j = i
while j < len(content):
if content[j] == "\\":
j += 2
continue
if content[j] == '"':
return i, j
j += 1
return i, len(content)
def latex_token_probs(content: str, items: list[dict]) -> tuple[float | None, float | None]:
"""(min, geometric mean) of the probabilities of the tokens that make up the LaTeX value.
*items* is the OpenAI `logprobs.content` list ({"token", "logprob"}). Token texts are laid end
to end to find which tokens overlap the LaTeX string; if they do not reproduce the answer
(some servers report bytes), all tokens are used.
"""
lps = [(it.get("token", ""), it.get("logprob")) for it in items if it.get("logprob") is not None]
if not lps:
return None, None
span = _latex_span(content)
joined = "".join(t for t, _ in lps)
sel: list[float] = []
if span and joined == content:
pos = 0
for tok, lp in lps:
a, b = pos, pos + len(tok)
pos = b
if b > span[0] and a < span[1] and tok.strip():
sel.append(float(lp))
if not sel:
sel = [float(lp) for t, lp in lps if t.strip()]
if not sel:
return None, None
return float(np.exp(min(sel))), float(np.exp(np.mean(sel)))
class VisionEngine:
def __init__(self, name: str, base_url: str, model: str, api_key: str = "", needs_key: bool = False,
thinking: bool | None = None) -> None:
self.name = name
self.thinking = thinking # None: server default; False: ask llama.cpp chat templates not to "think"
self.base_url = base_url.rstrip("/")
self.model = model
self.api_key = api_key
self.needs_key = needs_key
self.session = requests.Session()
def check(self) -> None:
"""Raise FormulaError if the endpoint is unusable; fills in the model name of a local server."""
if self.needs_key and not self.api_key:
raise FormulaError("OPENROUTER_API_KEY ist nicht gesetzt (für --engine cloud nötig)", EXIT_TOOL)
if not self.model:
try:
r = self.session.get(f"{self.base_url}/models", headers=self._headers(), timeout=10)
r.raise_for_status()
self.model = r.json()["data"][0]["id"]
except (requests.RequestException, KeyError, IndexError, ValueError) as e:
raise FormulaError(f"Lokaler Modellserver unter {self.base_url} nicht erreichbar ({e}). "
"Server starten (siehe BEDIENUNGSANLEITUNG) oder --engine cloud verwenden.",
EXIT_TOOL) from e
def _thinking_param(self) -> dict:
if self.thinking is None:
return {}
if "openrouter.ai" in self.base_url:
return {"reasoning": {"enabled": self.thinking}}
return {"chat_template_kwargs": {"enable_thinking": self.thinking}}
def _headers(self) -> dict[str, str]:
h = {"Content-Type": "application/json"}
if self.api_key:
h["Authorization"] = f"Bearer {self.api_key}"
if "openrouter.ai" in self.base_url:
h["X-Title"] = "formula_img2latex"
return h
def recognize(self, data: bytes, name: str = "", context: str = "", display: bool = False) -> Recognition:
png = prepare_image(data, name)
ctx = ""
if context:
where = "displayed after" if display else "part of"
ctx = CONTEXT_LINE.format(where=where, text=context.replace('"', "'")[:240])
body = {
"model": self.model,
"temperature": 0,
"max_tokens": MAX_TOKENS,
"logprobs": True,
"top_logprobs": 1,
**self._thinking_param(),
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": PROMPT.replace("{context}", ctx)},
{"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(png).decode()}},
],
}],
}
t0 = time.time()
last_err = ""
for attempt in range(3):
try:
r = self.session.post(f"{self.base_url}/chat/completions", headers=self._headers(),
json=body, timeout=TIMEOUT_S)
if r.status_code in (429, 500, 502, 503, 504):
last_err = f"HTTP {r.status_code}"
time.sleep(2 * (attempt + 1))
continue
r.raise_for_status()
js = r.json()
choice = js["choices"][0]
content = choice["message"].get("content") or ""
text = content.strip()
usage = js.get("usage") or {}
tex, isf = parse_answer(text)
p_min, p_mean = latex_token_probs(content, ((choice.get("logprobs") or {}).get("content") or []))
return Recognition(self.name, js.get("model", self.model), tex, isf, text[:2000], None,
time.time() - t0, int(usage.get("prompt_tokens") or 0),
int(usage.get("completion_tokens") or 0), p_min, p_mean, parse_confidence(text))
except (requests.RequestException, KeyError, IndexError, ValueError) as e:
last_err = str(e)[:300]
time.sleep(1 + attempt)
return Recognition(self.name, self.model, None, None, "", last_err or "unbekannter Fehler", time.time() - t0)