Test whether a vision model reads the right fields

import argparse
import base64
import json
from pathlib import Path

from semantic_notes import api

FIELDS = ("time", "room", "signup")
EXPECTED = {"time": "14:30", "room": "204", "signup": "front desk"}
SCHEMA = {
    "type": "object",
    "properties": {field: {"type": ["string", "null"]} for field in FIELDS},
    "required": list(FIELDS),
    "additionalProperties": False,
}

parser = argparse.ArgumentParser(description="Check a model's extraction on a workshop notice.")
parser.add_argument("image", type=Path)
parser.add_argument("--model", default="gemma3:4b")
parser.add_argument("--hidden", choices=FIELDS, help="Field covered in your edited image")
args = parser.parse_args()

try:
    image = base64.b64encode(args.image.read_bytes()).decode("ascii")
    response = api("/api/chat", {
        "model": args.model,
        "messages": [{
            "role": "user",
            "content": (
                "Transcribe three fields from the notice. Set time to the exact time, "
                "room to the exact room number, and signup to the place name only, "
                "with no leading preposition or article. "
                "Use null for a field you cannot read. Return the fields as JSON."
            ),
            "images": [image],
        }],
        "format": SCHEMA,
        "options": {"temperature": 0, "num_ctx": 4096, "num_predict": 128},
        "stream": False,
    })
    result = json.loads(response["message"]["content"])
    if not isinstance(result, dict) or set(result) != set(FIELDS):
        raise ValueError("The model did not return all three fields.")
except (OSError, KeyError, TypeError, ValueError) as error:
    raise SystemExit(f"Could not score the image: {error}") from None

correct = 0
for field in FIELDS:
    expected = None if field == args.hidden else EXPECTED[field]
    actual = result[field]
    if isinstance(actual, str):
        actual = " ".join(actual.casefold().strip(" .").split())
    passed = actual == expected
    correct += passed
    print(f"{'PASS' if passed else 'MISS'} | {field}: {actual!r} | expected {expected!r}")
print(f"{correct}/{len(FIELDS)} matched")

import argparse
from contextlib import closing
from itertools import islice
import json
import math
from pathlib import Path
import sqlite3
from urllib.error import HTTPError
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

BASE = "http://127.0.0.1:11434"
MODEL = "all-minilm:latest"
MAX_PASSAGES = 200


class NoRedirects(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def api(endpoint, payload=None):
    request = Request(BASE + endpoint)
    if payload is not None:
        request.data = json.dumps(payload).encode("utf-8")
        request.add_header("Content-Type", "application/json")
    opener = build_opener(ProxyHandler({}), NoRedirects())
    with opener.open(request, timeout=120) as response:
        return json.load(response)


def model_digest():
    for model in api("/api/tags")["models"]:
        if model["name"] == MODEL:
            return model["digest"]
    raise ValueError(f"Pull {MODEL} with Ollama first.")


def embed(texts):
    result = api("/api/embed", {
        "model": MODEL,
        "input": texts,
        "truncate": False,
    })["embeddings"]
    if len(result) != len(texts):
        raise ValueError("Ollama returned the wrong number of embeddings.")
    return result


def cosine(left, right):
    if len(left) != len(right):
        raise ValueError("Vector dimensions changed. Rebuild the index.")
    length = math.hypot(*left) * math.hypot(*right)
    return sum(x * y for x, y in zip(left, right)) / length if length else 0


def build_index(folder, database):
    from index_notes import passages

    rows = list(islice(passages(folder), MAX_PASSAGES + 1))
    if not rows or len(rows) > MAX_PASSAGES:
        raise ValueError(f"Use 1–{MAX_PASSAGES} nonempty passages for this example.")
    digest = model_digest()
    stored = []
    for start in range(0, len(rows), 16):
        batch = rows[start:start + 16]
        vectors = embed([text for _, _, text in batch])
        stored.extend((*row, json.dumps(vector)) for row, vector in zip(batch, vectors))
    if model_digest() != digest:
        raise ValueError("The embedding model changed during indexing. Build the index again.")
    with closing(sqlite3.connect(database)) as connection, connection:
        connection.execute("CREATE TABLE IF NOT EXISTS model (name TEXT, digest TEXT)")
        connection.execute(
            "CREATE TABLE IF NOT EXISTS passages "
            "(source TEXT, number INTEGER, text TEXT, vector TEXT)"
        )
        connection.execute("DELETE FROM passages")
        connection.execute("DELETE FROM model")
        connection.execute("INSERT INTO model VALUES (?, ?)", (MODEL, digest))
        connection.executemany("INSERT INTO passages VALUES (?, ?, ?, ?)", stored)
    print(f"Saved {len(stored)} embeddings in {database}.")


def find_notes(database, query):
    if not query.strip() or not database.is_file():
        raise ValueError("Build the index and enter a nonempty question.")
    with closing(sqlite3.connect(database.resolve().as_uri() + "?mode=ro", uri=True)) as connection:
        connection.execute("BEGIN")
        identity = connection.execute("SELECT name, digest FROM model").fetchone()
        rows = connection.execute("SELECT source, number, text, vector FROM passages").fetchall()
    digest = model_digest()
    if identity != (MODEL, digest):
        raise ValueError("The embedding model differs from the one in the index. Rebuild it.")
    vector, = embed([query])
    if model_digest() != digest:
        raise ValueError("The embedding model changed during the query. Rebuild the index.")
    matches = [
        (cosine(json.loads(embedding), vector), source, number, text)
        for source, number, text, embedding in rows
    ]
    ranked = sorted(matches, key=lambda match: match[0], reverse=True)
    for score, source, number, text in ranked[:3]:
        print(f"{score:.3f} | {source}, passage {number}\n{text}\n")


def main():
    parser = argparse.ArgumentParser(description="Store local embeddings and search them later.")
    parser.add_argument("--db", type=Path, default=Path("notes-vectors.sqlite3"))
    commands = parser.add_subparsers(dest="command", required=True)
    build = commands.add_parser("build")
    build.add_argument("folder", type=Path)
    find = commands.add_parser("find")
    find.add_argument("query")
    args = parser.parse_args()

    try:
        if args.command == "build":
            build_index(args.folder, args.db)
        else:
            find_notes(args.db, args.query)
    except HTTPError as error:
        raise SystemExit(f"Ollama returned HTTP {error.code}: {error.read().decode()}") from None
    except (OSError, sqlite3.Error, ValueError) as error:
        raise SystemExit(str(error)) from None


if __name__ == "__main__":
    main()