Antigravity で Gemma 4 のローカル出力に型安全を取り戻す — GBNF / Outlines / JSON Schema 制約デコードの使い分けを本番設計する

root ::= "{" ws "\"label\"" ws ":" ws label ws "}"
label ::= "\"billing\"" | "\"support\"" | "\"general\""
ws ::= [ \t\n]*

./llama-cli \
  -m ./models/gemma-4-9b-it-Q4_K_M.gguf \
  --grammar-file ./classify.gbnf \
  -p "Classify this message: I want a refund for order #1234" \
  -n 64 --temp 0.1

"""Gemma 4 + llama.cpp + GBNF で意図分類を行う最小実装。
 
前提: pip install llama-cpp-python==0.3.5
モデル: gemma-4-9b-it-Q4_K_M.gguf を ./models/ に配置済み
想定出力: {"label": "billing"} のような JSON
"""
from llama_cpp import Llama, LlamaGrammar
from typing import Literal
import json
 
GRAMMAR = r'''
root ::= "{" ws "\"label\"" ws ":" ws label ws "}"
label ::= "\"billing\"" | "\"support\"" | "\"general\""
ws ::= [ \t\n]*
'''
 
llm = Llama(
    model_path="./models/gemma-4-9b-it-Q4_K_M.gguf",
    n_ctx=4096,
    n_gpu_layers=-1,  # Apple Silicon なら全レイヤーを GPU に
    verbose=False,
)
grammar = LlamaGrammar.from_string(GRAMMAR)
 
def classify(message: str) -> Literal["billing", "support", "general"]:
    """顧客メッセージを3ラベルに分類する。制約デコードにより戻り値は保証される。"""
    prompt = f"<start_of_turn>user\nClassify: {message}<end_of_turn>\n<start_of_turn>model\n"
    resp = llm(
        prompt,
        grammar=grammar,
        max_tokens=32,
        temperature=0.1,
    )
    text = resp["choices"][0]["text"]
    data = json.loads(text)  # 文法で保証されているので 100% 成功する
    return data["label"]
 
if __name__ == "__main__":
    print(classify("I want a refund for order #1234"))  # => billing
    print(classify("Where is my order?"))               # => support
    print(classify("Hello, how are you?"))              # => general

root ::= "{" ws "\"intent\"" ws ":" ws value ws "}"
value ::= "\"" char* "\""
char ::= [^"\\] | "\\" (["\\/bfnrt] | "u" [0-9a-fA-F]{4})
ws ::= [ \t\n]*

"""Outlines + Gemma 4 でツール引数を型安全に抽出する。
 
前提: pip install outlines==0.1.11 transformers==4.45.0 torch
用途: エージェントが自然文リクエストからツール呼び出し引数を生成するケース
"""
import outlines
from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime
 
class MeetingRequest(BaseModel):
    """会議設定リクエストの構造化表現。"""
    attendee: str = Field(..., description="相手の名前")
    start_iso: str = Field(..., description="開始時刻 ISO 8601")
    duration_minutes: int = Field(..., ge=15, le=480)
    priority: Literal["low", "medium", "high"] = "medium"
 
model = outlines.models.transformers("google/gemma-4-9b-it", device="cuda")
generator = outlines.generate.json(model, MeetingRequest)
 
def extract_meeting(user_text: str) -> MeetingRequest:
    """自然文から会議リクエストを抽出する。失敗しうるのは Pydantic バリデーションのみ。"""
    prompt = f"""次のリクエストを構造化データに変換してください。
今日は 2026-04-24 です。
リクエスト: {user_text}
"""
    # generator の戻り値は MeetingRequest インスタンス。型が保証される。
    return generator(prompt, max_tokens=256)
 
if __name__ == "__main__":
    result = extract_meeting("明日の15時から鈴木さんと1時間、緊急で打ち合わせしたい")
    print(result.model_dump_json(indent=2))
    # {
    #   "attendee": "鈴木",
    #   "start_iso": "2026-04-25T15:00:00",
    #   "duration_minutes": 60,
    #   "priority": "high"
    # }

"""多段エージェントの各段で Outlines を使い分ける例。"""
from enum import Enum
from pydantic import BaseModel
import outlines
 
class Intent(str, Enum):
    SCHEDULE = "schedule"
    CANCEL = "cancel"
    QUERY = "query"
 
class IntentClassification(BaseModel):
    intent: Intent
    confidence: float
 
class CancelRequest(BaseModel):
    meeting_id: str
    reason: str
 
model = outlines.models.transformers("google/gemma-4-9b-it")
intent_gen = outlines.generate.json(model, IntentClassification)
cancel_gen = outlines.generate.json(model, CancelRequest)
 
def agent_pipeline(user_text: str) -> dict:
    """意図を分類し、その結果に応じて適切な引数抽出を行う。"""
    intent_result = intent_gen(f"Classify intent: {user_text}")
 
    if intent_result.confidence < 0.6:
        return {"status": "clarify_needed", "confidence": intent_result.confidence}
 
    if intent_result.intent == Intent.CANCEL:
        cancel = cancel_gen(f"Extract cancel info: {user_text}")
        return {"status": "ok", "action": "cancel", "data": cancel.model_dump()}
    # 他の意図も同様に分岐 ...
    return {"status": "ok", "intent": intent_result.intent.value}

python -m vllm.entrypoints.openai.api_server \
  --model google/gemma-4-9b-it \
  --dtype bfloat16 \
  --max-model-len 8192 \
  --port 8000 \
  --guided-decoding-backend outlines

"""vLLM サーバーを OpenAI SDK 経由で呼び出す。Antigravity のエージェント本番構成。
 
前提: pip install openai==1.55.0
vLLM サーバーが localhost:8000 で起動していること
"""
from openai import OpenAI
from pydantic import BaseModel
import json
 
class ProductReview(BaseModel):
    sentiment: str
    summary: str
    key_points: list[str]
    rating: int
 
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="EMPTY",  # vLLM はデフォルトで認証なし
)
 
def analyze_review(review_text: str) -> ProductReview:
    """商品レビューを構造化解析する。スキーマ違反はサーバー側で物理的に防止される。"""
    resp = client.chat.completions.create(
        model="google/gemma-4-9b-it",
        messages=[
            {"role": "system", "content": "You are a product review analyzer. Output JSON only."},
            {"role": "user", "content": review_text},
        ],
        extra_body={
            "guided_json": ProductReview.model_json_schema(),
        },
        temperature=0.3,
        max_tokens=512,
    )
    raw = resp.choices[0].message.content
    return ProductReview.model_validate_json(raw)
 
if __name__ == "__main__":
    review = analyze_review(
        "届いた商品は想像以上でした。梱包も丁寧で、配送も早く、価格も納得です。"
        "ただ色だけは写真と少し違った印象でした。"
    )
    print(review.model_dump_json(indent=2))

string ::= "\"" ( [^"\\] | "\\" (["\\/bfnrt] | "u" [0-9a-fA-F]{4}) )* "\""

"""JSON Schema の書き方ごとに, 制約デコードの索引がどれだけ重くなるかを測る.
 
依存なし (Python 3.10+ 標準ライブラリのみ).
スキーマ相当の正規表現を組み, NFA -> DFA と落として状態数と構築時間を測る.
"""
import time
import statistics
 
 
class R:
    """正規表現ノード. op は lit / seq / alt / opt / star."""
    __slots__ = ("op", "kids", "sym")
 
    def __init__(self, op, kids=(), sym=None):
        self.op, self.kids, self.sym = op, tuple(kids), sym
 
 
def lit(sym): return R("lit", (), sym)
def opt(k):   return R("opt", (k,))
def star(k):  return R("star", (k,))
def text(s):  return seq(*[lit(c) for c in s])
 
 
def seq(*ks):
    ks = [k for k in ks if k is not None]
    return ks[0] if len(ks) == 1 else R("seq", ks)
 
 
def alt(*ks):
    return ks[0] if len(ks) == 1 else R("alt", ks)
 
 
# アルファベットは実文字 + 「その他の印字可能文字」1 個で近似する
STRUCT, DIGITS = set('{}[]:,"\\ '), set("0123456789")
OTHER = "\x00"
ALPHABET = STRUCT | DIGITS | set("abcdefghijklmnopqrstuvwxyz_-") | {OTHER}
 
 
def free_string():
    """任意の JSON 文字列. エスケープも受理する."""
    body = star(alt(alt(*[lit(c) for c in sorted(ALPHABET - set('"\\'))]),
                    seq(lit('\\'), alt(lit('"'), lit('\\'), lit('n'), lit('t')))))
    return seq(lit('"'), body, lit('"'))
 
 
def integer():
    d = alt(*[lit(c) for c in sorted(DIGITS)])
    return seq(opt(lit('-')), d, star(d))
 
 
def enum_of(words):
    return alt(*[text('"%s"' % w) for w in words])
 
 
def obj(pairs, additional=False):
    """キーはスキーマ順に出る前提 (Outlines の既定挙動に合わせる)."""
    inner = []
    for i, (k, v) in enumerate(pairs):
        if i:
            inner.append(lit(','))
        inner.append(seq(text('"%s"' % k), lit(':'), v))
    body = seq(*inner)
    if additional:                      # additionalProperties: true 相当
        body = seq(body, star(seq(lit(','), free_string(), lit(':'),
                                  alt(free_string(), integer()))))
    return seq(lit('{'), body, lit('}'))
 
 
def array(item, lo, hi):
    """minItems=lo, maxItems=hi の配列は内部で展開される."""
    parts = []
    for i in range(lo):
        if i:
            parts.append(lit(','))
        parts.append(item)
    for i in range(hi - lo):
        parts.append(opt(seq(lit(','), item) if (lo + i) > 0 else item))
    return seq(lit('['), seq(*parts) if parts else None, lit(']'))
 
 
class NFA:
    def __init__(self):
        self.t, self.e = [], []         # t: 文字遷移, e: ε 遷移
 
    def new(self):
        self.t.append({})
        self.e.append([])
        return len(self.t) - 1
 
 
def build(node, n, s):
    """node を NFA に展開し, 受理側の状態番号を返す."""
    if node.op == "lit":
        d = n.new()
        n.t[s].setdefault(node.sym, set()).add(d)
        return d
    if node.op == "seq":
        for k in node.kids:
            s = build(k, n, s)
        return s
    if node.op == "alt":
        out = n.new()
        for k in node.kids:
            b = n.new()
            n.e[s].append(b)
            n.e[build(k, n, b)].append(out)
        return out
    if node.op == "opt":
        out = n.new()
        n.e[s].append(out)
        b = n.new()
        n.e[s].append(b)
        n.e[build(node.kids[0], n, b)].append(out)
        return out
    if node.op == "star":
        loop = n.new()
        n.e[s].append(loop)
        b = n.new()
        n.e[loop].append(b)
        n.e[build(node.kids[0], n, b)].append(loop)
        return loop
    raise ValueError(node.op)
 
 
def closure(n, states):
    """ε 閉包."""
    stack, seen = list(states), set(states)
    while stack:
        for d in n.e[stack.pop()]:
            if d not in seen:
                seen.add(d)
                stack.append(d)
    return frozenset(seen)
 
 
def dfa_states(node):
    """部分集合構成で DFA の状態数を数える."""
    n = NFA()
    start = n.new()
    build(node, n, start)
    i0 = closure(n, {start})
    seen, work = {i0}, [i0]
    while work:
        moves = {}
        for s in work.pop():
            for sym, ds in n.t[s].items():
                moves.setdefault(sym, set()).update(ds)
        for ds in moves.values():
            nxt = closure(n, ds)
            if nxt not in seen:
                seen.add(nxt)
                work.append(nxt)
    return len(seen)
 
 
def measure(node, reps=5):
    ts = []
    for _ in range(reps):
        t0 = time.perf_counter()
        states = dfa_states(node)
        ts.append((time.perf_counter() - t0) * 1000)
    return states, statistics.median(ts)
 
 
LABELS = ["billing", "support", "general", "refund", "shipping", "account", "legal",
          "abuse", "spam", "other", "sales", "press", "bug", "feature", "docs",
          "trial", "renew", "cancel", "upsell", "churn"]
 
 
def nested(depth, leaf):
    node = leaf
    for _ in range(depth - 1):
        node = obj([("child", node)])
    return node
 
 
if __name__ == "__main__":
    print("[1] フィールド 1 本のコスト (int vs 自由文字列)")
    for n in (1, 2, 3, 4, 5):
        a = measure(obj([("f%d" % i, integer()) for i in range(n)]))
        b = measure(obj([("f%d" % i, free_string()) for i in range(n)]))
        print(f"  fields={n}  int: {a[0]:4d} states {a[1]:7.2f}ms | "
              f"str: {b[0]:4d} states {b[1]:8.2f}ms")
 
    print("[2] 入れ子の深さ (enum + int のみ, 自由文字列なし)")
    leaf = obj([("label", enum_of(LABELS[:3])), ("score", integer())])
    for d in (1, 3, 5, 8):
        s, ms = measure(nested(d, leaf))
        print(f"  depth={d}  {s:4d} states {ms:7.2f}ms")
 
    print("[3] 開いた口・配列の上限・enum の幅")
    fields = [("label", enum_of(LABELS[:3])), ("note", free_string()),
              ("score", integer())]
    for name, node in (("additionalProperties:false", obj(fields)),
                       ("additionalProperties:true ", obj(fields, additional=True))):
        s, ms = measure(node)
        print(f"  {name}  {s:4d} states {ms:8.2f}ms")
    for lo, hi in ((1, 1), (1, 3), (1, 10), (1, 20)):
        s, ms = measure(array(enum_of(LABELS[:3]), lo, hi))
        print(f"  array items={lo}..{hi:<2d}        {s:4d} states {ms:8.2f}ms")
    for e in (3, 5, 10, 20):
        s, ms = measure(obj([("label", enum_of(LABELS[:e])), ("note", free_string())]))
        print(f"  enum={e:<2d} choices           {s:4d} states {ms:8.2f}ms")

"""分類器の CI テスト. 温めたモデルに対して数秒で終わる."""
import pytest
 
from agents.classifier import classify
 
 
@pytest.mark.parametrize("message,expected", [
    ("I want a refund for order #1234", "billing"),
    ("Where is my order?", "support"),
    ("Hello, how are you?", "general"),
])
def test_classify(message: str, expected: str) -> None:
    assert classify(message) == expected

"""スキーマのスナップショット比較. 差分が出たら人が確認する."""
import json
from pathlib import Path
 
from agents.schemas import MeetingRequest
 
SNAPSHOT = Path(__file__).parent / "snapshots" / "meeting_request.schema.json"
 
 
def test_schema_unchanged() -> None:
    current = MeetingRequest.model_json_schema()
    expected = json.loads(SNAPSHOT.read_text(encoding="utf-8"))
    assert current == expected, (
        "スキーマが変わりました. 意図した変更なら snapshot を更新してください."
    )