Structured Output - 文法で次のトークンを絞って JSON の形を守らせる

"""応答の 1 トークン目について、モデルが付けた確率の上位を Ollama から取り出す（冒頭の図の元データ）。

format を付けずに、chat と同じ形のプロンプトを raw で /api/generate に送り、
logprobs と top_logprobs で 1 トークン目の候補の確率を受け取る。
各候補が、スキーマから作った文法の先頭（"{" の後に空白、続けて "person" のキー）に合うかも判定する。

使い方:
    OLLAMA_HOST=127.0.0.1:11435 python3 probe.py <モデル> <入力の番号> [出力の JSON]
"""
import json
import math
import os
import re
import sys
import urllib.request

from run import PROMPT
from texts import make_items

# 文法が 1 トークン目に許す文字列: "{"、空白、'"person"' の先頭部分の順
ALLOWED = re.compile(r'^\{[ \t\n]*(?:"(?:p(?:e(?:r(?:s(?:o(?:n(?:")?)?)?)?)?)?)?)?$')


def main():
    model, idx = sys.argv[1], int(sys.argv[2])
    out_path = sys.argv[3] if len(sys.argv) > 3 else None
    host = os.environ.get("OLLAMA_HOST", "127.0.0.1:11434")
    text = make_items()[idx]["text"]
    # qwen3 のチャットテンプレートで think: false を指定した時と同じ並び
    # （テンプレートは最後のユーザー発言の末尾に " /no_think" を足す。
    #   /api/chat に "_debug_render_only": true を付けて描画した結果と照合した）
    prompt = ("<|im_start|>user\n" + PROMPT.format(text=text) + " /no_think<|im_end|>\n"
              "<|im_start|>assistant\n<think>\n\n</think>\n\n")
    body = {"model": model, "prompt": prompt, "raw": True, "stream": False,
            "logprobs": True, "top_logprobs": 10,
            "options": {"temperature": 0, "seed": 42, "num_predict": 1}}
    req = urllib.request.Request(f"http://{host}/api/generate", data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=900) as r:
        res = json.load(r)
    print(f"input {idx}: {text}")
    print(f"greedy first token: {res['response']!r}")
    cands = []
    for t in res["logprobs"][0]["top_logprobs"]:
        p = math.exp(t["logprob"])
        ok = bool(ALLOWED.match(t["token"]))
        cands.append({"token": t["token"], "p": p, "allowed": ok,
                      "shown": t["token"].replace("\n", "\\n")})
        print(f"  {t['token']!r:>12}  p={p:.4f}  {'allowed' if ok else 'rejected'}")
    if out_path:
        with open(out_path, "w") as f:
            json.dump({"model": model, "input": idx, "candidates": cands,
                       "prefix_shown": "1 トークン目",
                       "caption": f"点線はスキーマの文法に合わない候補（{model}、入力 {idx}）"},
                      f, indent=2, ensure_ascii=False)


if __name__ == "__main__":
    main()

"""抽出課題の入力文と正解を作る。文は全て本記事のために作った架空のもので、人物も実在しない。

テンプレートに名前・日付・金額を差し込むので、正解は作る時点で決まる。
日付と金額は、書き方（西暦・スラッシュ・和暦、カンマ・万・円記号）を変えて混ぜる。
"""
import random

NAMES = ["佐藤花子", "鈴木一郎", "高橋健太", "田中美咲", "伊藤翔", "渡辺由美",
         "山本大輔", "中村さくら", "小林拓也", "加藤陽子", "吉田誠", "松本彩"]

TEMPLATES = [
    "{date}、{name}さんは書店で技術書を3冊、合計{amt}で購入した。",
    "{name}様より、{date}付けで{amt}のご入金を確認いたしました。請求書番号は INV-{num} です。",
    "経費精算 申請者: {name} / 利用日: {date} / 金額: {amt}（タクシー代、乗車人数2名）",
    "{date}に開かれた勉強会で、幹事の{name}さんが参加者12名分の会場費{amt}を立て替えた。",
    "山田さんの代わりに、{name}さんが{date}に家賃{amt}を振り込んだ。部屋番号は204号室。",
    "{name}さんから{date}の午後3時に連絡があり、修理代として{amt}を支払うとのことだった。",
    "Receipt: {date} 支払者 {name} 合計 {amt}",
    "{date}締めの請求として、{name}さんに{amt}（税込、送料500円を含む）を請求した。",
]


def date_text(rng, y, m, d):
    kind = rng.choice(["kanji", "slash", "wareki", "kanji"])
    if kind == "kanji":
        return f"{y}年{m}月{d}日"
    if kind == "slash":
        return f"{y}/{m}/{d}"
    return f"令和{y - 2018}年{m}月{d}日"  # 2019 年が令和元年


def amount_text(rng, a):
    kinds = ["comma", "yen_sign", "plain"]
    if a >= 10000 and a % 1000 == 0:
        kinds += ["man", "man"]
    if a >= 10000 and a % 10000 == 5000:
        kinds.append("man_decimal")
    kind = rng.choice(kinds)
    if kind == "comma":
        return f"{a:,}円"
    if kind == "yen_sign":
        return f"¥{a:,}"
    if kind == "plain":
        return f"{a}円"
    if kind == "man_decimal":
        return f"{a / 10000:g}万円"
    man, rest = divmod(a, 10000)
    sen = rest // 1000
    return f"{man}万" + (f"{sen}千" if sen else "") + "円"


def make_items(n=40, seed=20261110):
    rng = random.Random(seed)
    items = []
    for i in range(n):
        name = rng.choice(NAMES)
        y, m, d = rng.choice([2024, 2025, 2026]), rng.randint(1, 12), rng.randint(1, 28)
        a = rng.choice([980, 1500, 3200, 4500, 12800, 15000, 25000, 32000, 48000,
                        65000, 120000, 135000])
        tpl = TEMPLATES[i % len(TEMPLATES)]
        text = tpl.format(name=name, date=date_text(rng, y, m, d),
                          amt=amount_text(rng, a), num=rng.randint(1000, 9999))
        items.append({"id": i, "text": text, "answer": {
            "person": name, "date": f"{y:04d}-{m:02d}-{d:02d}", "amount_yen": a}})
    return items


if __name__ == "__main__":
    for it in make_items():
        print(it["id"], it["text"], it["answer"])

"""同じ抽出課題を 3 つの方式で Ollama の /api/chat に解かせ、応答と時間を JSON Lines に保存する。

方式:
  prompt : プロンプトで JSON を頼むだけ（format を付けない）
  json   : format: "json"（JSON の構文だけを守らせる）
  schema : format: <JSON Schema>（スキーマから作った文法で生成を縛る）
  schema_bare : schema と同じ format で、プロンプトにはキーも形式も書かない（追加の比較）

使い方:
    OLLAMA_HOST=127.0.0.1:11435 python3 run.py <出力の .jsonl> <モデル> <繰り返し回数> [方式,...]
方式の既定は prompt,json,schema。
1 回目は 40 件全て、2 回目以降は時間を測るために先頭の 20 件だけを解かせる。
出力に既にある組（モデル・回・入力・方式）は飛ばすので、途中で止めても続きから再開できる。
Python 標準ライブラリだけで動く。
"""
import json
import os
import sys
import time
import urllib.request

from texts import make_items

SCHEMA = {
    "type": "object",
    "properties": {
        "person": {"type": "string"},
        "date": {"type": "string", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$"},
        "amount_yen": {"type": "integer"},
    },
    "required": ["person", "date", "amount_yen"],
    "additionalProperties": False,
}
FORMATS = {"prompt": None, "json": "json", "schema": SCHEMA, "schema_bare": SCHEMA}
N_ITEMS, TIMING_ITEMS = 40, 20

PROMPT = """次の文から、取引の当事者（お金を払った・受け取った・請求された人）の名前、日付、金額を取り出してください。
出力は JSON だけにしてください。キーは person（敬称を付けない名前）、date（YYYY-MM-DD 形式）、amount_yen（円単位の整数）の 3 つです。

文: {text}"""

# schema_bare 用: 何を取り出すかだけを書き、キー名・日付の形式・JSON で答えることは書かない
PROMPT_BARE = """次の文から、取引の当事者の名前、日付、金額を取り出してください。

文: {text}"""


def chat(host, model, text, fmt, prompt=PROMPT):
    body = {
        "model": model,
        "messages": [{"role": "user", "content": prompt.format(text=text)}],
        "stream": False,
        "think": False,  # qwen3 の思考モードを切る（qwen3:4b は切れない。本文参照）
        "options": {"temperature": 0, "seed": 42, "num_predict": 512},
    }
    if fmt is not None:
        body["format"] = fmt
    req = urllib.request.Request(f"http://{host}/api/chat", data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"})
    t0 = time.perf_counter()
    with urllib.request.urlopen(req, timeout=1800) as r:
        res = json.load(r)
    res["wall_s"] = time.perf_counter() - t0
    return res


def main():
    out_path, model, repeats = sys.argv[1], sys.argv[2], int(sys.argv[3])
    names = sys.argv[4].split(",") if len(sys.argv) > 4 else ["prompt", "json", "schema"]
    host = os.environ.get("OLLAMA_HOST", "127.0.0.1:11434")
    items = make_items(N_ITEMS)
    done = set()
    if os.path.exists(out_path):
        for line in open(out_path):
            r = json.loads(line)
            done.add((r["model"], r["rep"], r["id"], r["method"]))
    with open(out_path, "a") as f:
        for rep in range(repeats):
            for it in items if rep == 0 else items[:TIMING_ITEMS]:
                k = (it["id"] + rep) % len(names)  # 方式の順番を回して、負荷の偏りを散らす
                for method in names[k:] + names[:k]:
                    if (model, rep, it["id"], method) in done:
                        continue
                    prompt = PROMPT_BARE if method == "schema_bare" else PROMPT
                    res = chat(host, model, it["text"], FORMATS[method], prompt)
                    f.write(json.dumps({
                        "model": model, "rep": rep, "id": it["id"], "method": method,
                        "content": res["message"]["content"],
                        "thinking": res["message"].get("thinking", ""),
                        "done_reason": res.get("done_reason"),
                        "eval_count": res.get("eval_count"),
                        "eval_s": res.get("eval_duration", 0) / 1e9,
                        "prompt_eval_s": res.get("prompt_eval_duration", 0) / 1e9,
                        "load_s": res.get("load_duration", 0) / 1e9,
                        "total_s": res.get("total_duration", 0) / 1e9,
                        "wall_s": res["wall_s"],
                    }, ensure_ascii=False) + "\n")
                    f.flush()
            print(f"{model} rep {rep} done", flush=True)


if __name__ == "__main__":
    main()

"""run.py の結果（JSON Lines）を、パース・スキーマ・値の 3 段で採点し、時間をまとめる。

使い方:
    uv run --with jsonschema==4.25.1 python3 evaluate.py <結果の .jsonl>... [集計の .json]
"""
import json
import statistics
import sys
from collections import defaultdict

import jsonschema

from run import N_ITEMS, SCHEMA, TIMING_ITEMS
from texts import make_items

METHODS = ["prompt", "json", "schema", "schema_bare"]


def score(content, answer):
    """3 段の判定。前の段を通らなければ後の段は False。"""
    try:
        obj = json.loads(content)
    except json.JSONDecodeError:
        return {"parse": False, "schema": False, "value": False, "fields": {}}
    ok_schema = jsonschema.Draft202012Validator(SCHEMA).is_valid(obj)
    fields = {k: ok_schema and obj.get(k) == v for k, v in answer.items()}
    return {"parse": True, "schema": ok_schema, "value": all(fields.values()),
            "fields": fields}


def fenced(content):
    """```json ... ``` で囲まれていて、囲みを外せばパースできるか。"""
    t = content.strip()
    if not t.startswith("```"):
        return False
    t = t.split("\n", 1)[-1].rsplit("```", 1)[0]
    try:
        json.loads(t)
        return True
    except json.JSONDecodeError:
        return False


def main():
    paths = [a for a in sys.argv[1:] if a.endswith(".jsonl")]
    outs = [a for a in sys.argv[1:] if a.endswith(".json")]
    out_path = outs[0] if outs else None
    answers = {it["id"]: it["answer"] for it in make_items(N_ITEMS)}
    rows = [json.loads(line) for path in paths for line in open(path)]
    models = list(dict.fromkeys(r["model"] for r in rows))
    summary = {}
    for model in models:
        print(f"[{model}]")
        print(f"  {'method':<11} {'fence':>5} {'parse':>6} {'schema':>7} {'value':>6}"
              f" {'person':>7} {'date':>5} {'amount':>7}  {'tokens':>6} {'sec/call':>8}"
              f"  {'tok/s':>5}  {'rep spread':>10}  same")
        summary[model] = {}
        for method in METHODS:
            rs = [r for r in rows if r["model"] == model and r["method"] == method]
            if not rs:
                continue
            first = [r for r in rs if r["rep"] == 0]
            n = len(first)
            sc = [score(r["content"], answers[r["id"]]) for r in first]
            cnt = {k: sum(s[k] for s in sc) for k in ("parse", "schema", "value")}
            cnt["fenced"] = sum(fenced(r["content"]) for r in first)
            fld = {k: sum(s["fields"].get(k, False) for s in sc)
                   for k in ("person", "date", "amount_yen")}
            # 時間: 他のモデルとの入れ替えで毎回読み込みが入るので、total_duration ではなく
            # プロンプトの処理と生成の時間（prompt_eval + eval）を使う。
            # 繰り返しごとに 1 呼び出しの中央値を出し、その中央値と幅を見る
            # 繰り返したのは先頭の TIMING_ITEMS 件だけなので、時間はその件数で比べる
            timed = [r for r in rs if r["id"] < TIMING_ITEMS]
            per_rep = defaultdict(list)
            for r in timed:
                per_rep[r["rep"]].append(r)
            med = [statistics.median(x["prompt_eval_s"] + x["eval_s"] for x in v)
                   for v in per_rep.values()]
            tps = statistics.median(r["eval_count"] / r["eval_s"] for r in timed if r["eval_s"])
            toks = statistics.median(r["eval_count"] for r in first)
            # 繰り返した入力のうち、応答の文字列が全ての回で同じだった件数
            by_id = defaultdict(set)
            for r in timed:
                by_id[r["id"]].add(r["content"])
            same = sum(len(v) == 1 for v in by_id.values())
            n_timed = len(by_id)
            spread = f"{min(med):.2f}-{max(med):.2f}" if len(med) > 1 else "-"
            same_s = f"{same}/{n_timed}" if len(med) > 1 else "-"
            print(f"  {method:<11} {cnt['fenced']:>5} {cnt['parse']:>3}/{n} {cnt['schema']:>4}/{n} {cnt['value']:>3}/{n}"
                  f" {fld['person']:>7} {fld['date']:>5} {fld['amount_yen']:>7}  {toks:>6.0f}"
                  f" {statistics.median(med):>8.2f}  {tps:>5.1f}  {spread:>10}  {same_s}")
            summary[model][method] = {"n": n, **cnt, **fld, "tokens": toks,
                                      "sec_median": statistics.median(med),
                                      "sec_reps": med, "tok_per_s": tps, "same": same, "n_timed": n_timed,
                                      "length_stop": sum(r["done_reason"] == "length" for r in first)}
        print()
    if out_path:
        with open(out_path, "w") as f:
            json.dump(summary, f, indent=2, ensure_ascii=False)


if __name__ == "__main__":
    main()

$ uv run --with jsonschema==4.25.1 python3 evaluate.py results.jsonl summary.json
[qwen3:1.7b-q4_K_M]
  method      fence  parse  schema  value  person  date  amount  tokens sec/call  tok/s  rep spread  same
  prompt         25  15/40   15/40  13/40      14    14      15      44     0.93   46.6   0.79-1.91  13/20
  json            0  40/40   40/40  31/40      36    35      40      37     0.76   48.6   0.72-1.46  20/20
  schema          0  40/40   40/40  31/40      36    35      40      37     0.77   49.2   0.69-1.66  20/20
  schema_bare     0  40/40   40/40  23/40      28    32      40      38     0.87   52.5           -  -
[qwen3:4b]
  method      fence  parse  schema  value  person  date  amount  tokens sec/call  tok/s  rep spread  same
  prompt          0   0/20    0/20   0/20       0     0       0     512    11.52   44.8           -  -
  json            0  20/20   20/20  18/20      20    18      20      42     1.16   42.3           -  -
  schema          0  20/20   20/20  18/20      20    18      20      42     0.82   56.6           -  -
[qwen3:8b]
  method      fence  parse  schema  value  person  date  amount  tokens sec/call  tok/s  rep spread  same
  prompt          0  40/40   40/40  36/40      40    38      37      42     1.84   23.8           -  -
  json            0  40/40   40/40  36/40      40    38      37      42     1.93   22.9           -  -
  schema          0  40/40   40/40  36/40      40    38      37      42     1.76   25.5           -  -
  schema_bare     0  40/40   40/40  24/40      27    36      40      42     1.82   25.4           -  -

$ python3 think.py qwen3:1.7b-q4_K_M 3; python3 think.py qwen3:4b 3
qwen3:1.7b-q4_K_M think=false format=none   thinking=    0 chars  tokens=  46  eval=   1.1 s  wall=  12.0 s  done=stop
    content: ```json { "person": "松本彩", "date": "2024-03-19", "amount_yen": 120000 
qwen3:1.7b-q4_K_M think=false format=schema thinking=    0 chars  tokens=  37  eval=   0.9 s  wall=   1.1 s  done=stop
    content: {"person": "松本彩", "date": "2024-03-19", "amount_yen": 120000}
qwen3:1.7b-q4_K_M think=true  format=none   thinking= 1168 chars  tokens= 463  eval=   7.4 s  wall=   7.7 s  done=stop
    content: { "person": "松本彩", "date": "2024-03-19", "amount_yen": 120000 }
qwen3:1.7b-q4_K_M think=true  format=schema thinking= 1116 chars  tokens=  42  eval=   1.7 s  wall=  13.6 s  done=stop
    content: { "person": "松本彩", "date": "2024-03-19", "amount_yen": 120000 }
qwen3:4b think=false format=none   thinking=    0 chars  tokens=2387  eval=  89.7 s  wall= 301.4 s  done=stop
    content: Okay, let's tackle this problem. So, I need to extract the names of th
    ...末尾: </think> {"person": "松本彩", "date": "2024-03-19", "amount_yen": 120000}
qwen3:4b think=false format=schema thinking=    0 chars  tokens=  42  eval=   1.3 s  wall=  16.5 s  done=stop
    content: { "person": "松本彩", "date": "2024-03-19", "amount_yen": 120000 }
qwen3:4b think=true  format=none   thinking= 7964 chars  tokens=2387  eval=  80.7 s  wall=  83.1 s  done=stop
    content: {"person": "松本彩", "date": "2024-03-19", "amount_yen": 120000}
qwen3:4b think=true  format=schema thinking= 7964 chars  tokens=  37  eval=   1.4 s  wall=  89.1 s  done=stop
    content: {"person": "松本彩", "date": "2024-03-19", "amount_yen": 120000}

"""think（思考モード）と format の組み合わせで、思考の文がどこに返るかを見る。

同じ入力 1 件を、think: false / true と format 無し / スキーマの組で解かせ、
content の先頭、thinking の長さ、生成したトークン数と時間を出す。
tokens と eval は Ollama の応答の eval_count と eval_duration、wall は要求を送ってから応答が返るまでの時間。

使い方:
    OLLAMA_HOST=127.0.0.1:11435 python3 think.py <モデル> <入力の番号> [think の値,...]
"""
import json
import os
import sys
import time
import urllib.request

from run import PROMPT, SCHEMA
from texts import make_items


def main():
    model, idx = sys.argv[1], int(sys.argv[2])
    thinks = [v == "true" for v in sys.argv[3].split(",")] if len(sys.argv) > 3 else [False, True]
    host = os.environ.get("OLLAMA_HOST", "127.0.0.1:11434")
    text = make_items()[idx]["text"]
    for think in thinks:
        for name, fmt in (("none", None), ("schema", SCHEMA)):
            body = {"model": model, "stream": False, "think": think,
                    "messages": [{"role": "user", "content": PROMPT.format(text=text)}],
                    "options": {"temperature": 0, "seed": 42, "num_predict": 4096}}
            if fmt is not None:
                body["format"] = fmt
            req = urllib.request.Request(f"http://{host}/api/chat",
                                         data=json.dumps(body).encode(),
                                         headers={"Content-Type": "application/json"})
            t0 = time.perf_counter()
            with urllib.request.urlopen(req, timeout=1800) as r:
                res = json.load(r)
            wall = time.perf_counter() - t0
            msg = res["message"]
            flat = " ".join(msg["content"].split())
            print(f"{model} think={str(think).lower():<5} format={name:<6} "
                  f"thinking={len(msg.get('thinking', '')):>5} chars  "
                  f"tokens={res['eval_count']:>4}  eval={res['eval_duration'] / 1e9:6.1f} s  "
                  f"wall={wall:6.1f} s  done={res['done_reason']}\n    content: {flat[:70]}")
            if len(flat) > 140:
                print(f"    ...末尾: {flat[-70:]}")


if __name__ == "__main__":
    main()

"""金額が書かれていない文を、スキーマで縛って解かせる（値を作ってしまうかを見る）。

amount_yen を必須の整数にしたスキーマ（run.py の SCHEMA）、null も許すスキーマ、
null も許した上でプロンプトに「書かれていない項目は null」と足した場合（integer|null+ask）の 3 つで、同じ文を解かせる。

使い方:
    OLLAMA_HOST=127.0.0.1:11435 python3 missing.py <モデル>...
"""
import copy
import os
import sys

from run import PROMPT, SCHEMA, chat

TEXT = "佐藤花子さんから2025年6月1日に、先月の請求書について問い合わせの電話があった。"

NULLABLE = copy.deepcopy(SCHEMA)
NULLABLE["properties"]["amount_yen"] = {"type": ["integer", "null"]}
PROMPT_NULL = PROMPT.replace("\n\n文:", "\n文に書かれていない項目は null にしてください。\n\n文:")


def main():
    host = os.environ.get("OLLAMA_HOST", "127.0.0.1:11434")
    print(f"text: {TEXT}")
    for model in sys.argv[1:]:
        for name, schema, prompt in (("integer", SCHEMA, PROMPT),
                                     ("integer|null", NULLABLE, PROMPT),
                                     ("integer|null+ask", NULLABLE, PROMPT_NULL)):
            res = chat(host, model, TEXT, schema, prompt)
            out = " ".join(res["message"]["content"].split())  # 改行と字下げを詰めて 1 行で表示
            print(f"{model:<18} {name:<17} -> {out}")


if __name__ == "__main__":
    main()