Skip to content
Published on

LLMの構造化出力を安定させる — パース失敗をなくす四重の防御

シェア
Authors

はじめに — パースは99パーセント成功しても失敗です

LLMの出力をパイプラインの中間段階として使い始めると、自然言語の品質よりパース成功率が先に問題になります。要約が少し味気ないのは我慢できますが、一日10万件のうち1パーセントがパースに失敗すれば、それは1千件の例外処理です。

この問題が厄介なのは、失敗が毎回違う姿で来るところにあります。昨日はマークダウンのフェンスに包まれて来て、今日は後ろに親切な説明文が付いて、明日は途中で切れます。そのたびに文字列処理を一つずつ足していくと、正規表現20行のパーサができあがり、それでも結局どこかから漏れます。

解決策は防御を層で積みつつ、各層が何を止めて何を止められないのかを明確に知ることです。この記事ではまず失敗の類型を分類し、そのうえで四層の防御を順に配置します。

まず失敗の類型を分類します

ログを開いてパース失敗のケースを100件だけ集めて分類すれば、対応の優先順位はすぐ決まります。おおむね六つに分かれます。

失敗の類型症状原因最も安い防御
コードフェンス包装バックティック三つとjson表記で包まれて来る学習データのマークダウンの習慣パース前にフェンスを除去
前後の説明文「ご依頼の結果です」のような前置き対話型アラインメントの副作用事前充填、停止シーケンス
途中で切れるオブジェクトが閉じないまま終わるトークン上限への到達終了理由の確認、上限の再算定
誤ったエスケープ文字列内の引用符と改行の処理誤り長い自由テキストフィールドフィールドの分離、制約デコーディング
幻覚フィールドスキーマにないキーが追加されるスキーマ制約の不在厳格スキーマ、未知のキーを拒否
型の不一致数値を文字列で、日付形式がまちまち型の明示不足フィールド名と説明に形式を明示

この表で重要なのは、最後の列がすべて違うという点です。一つの対策で全部を止めようとすると過剰設計になります。実際の分布を見て上位二つか三つを処理するだけでも、失敗率はたいてい一桁下に落ちます。

前の二つの類型は特に安く止まります。

import json, re

FENCE = re.compile(r"^\s*```(?:json)?\s*|\s*```\s*$", re.MULTILINE)

def extract_json(text):
    """1) フェンスを除去 2) それでも駄目なら最初のオブジェクト区間だけ切り出して試す"""
    cleaned = FENCE.sub("", text).strip()
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        pass
    start, end = cleaned.find("{"), cleaned.rfind("}")
    if start != -1 and end > start:
        return json.loads(cleaned[start:end + 1])
    raise ValueError("JSON을 찾을 수 없음")

これは応急処置です。根本的な対策は下の階層で扱いますが、この十行が今夜のアラートを止めてくれるのも事実です。

防御の四層 — 指示、例示、スキーマ、制約

防御手段を強度順に並べるとこうなります。上に行くほど強く、下に行くほど安くどこでも使えます。

第一層はプロンプトの指示です。「JSONだけを出力し、他のテキストは付けないでください」は驚くほどよく効きます。ここに二つを足すと効果が大きくなります。停止シーケンスを指定してオブジェクトが閉じた後に続く説明を切り落とし、アシスタントの応答を開き波括弧であらかじめ始めておく事前充填を使います。事前充填は前置きを物理的に不可能にします。

第二層は例示です。望む形を一つか二つ見せることが形式の遵守に与える効果は、指示文数行より大きいのです。特に列挙値、日付形式、空の値の表し方のように、言葉で説明すると冗長になる規則でそうなります。

第三層は関数呼び出し、あるいはツールスキーマです。自由テキストでJSONを要求する代わりに、引数スキーマを持つツールを定義してそのツールを呼ばせます。プロバイダ側でスキーマを強制してくれるので、先ほどの失敗類型のほとんどが消えます。APIをお使いなら、この層から始めるのが妥当です。

第四層は制約デコーディングです。スキーマを有限状態機械や文法にコンパイルし、毎ステップでその状態から許されないトークンの確率を0にします。出力が文法に違反することが原理的に不可能になります。

# オープンモデルを自分でサービングするなら制約デコーディングはサーバオプションで有効にします
vllm serve <모델경로> --guided-decoding-backend xgrammar

# リクエスト側からスキーマを渡します (例: OpenAI互換エンドポイント)
#   "guided_json": { ...JSON Schema... }
#   "guided_choice": ["肯定", "否定", "中立"]
#   "guided_regex": "\\d{4}-\\d{2}-\\d{2}"

分類のように出力がいくつかのラベルのうち一つである作業ではguided_choiceが特に強力です。出力空間が有限なので、パースという概念そのものが消えます。

制約デコーディングが保証するものと保証しないもの

ここで最も重要な区別をします。制約デコーディングは文法的な妥当性を保証します。意味的な正確性は保証しません

スキーマがpriceフィールドを整数として要求すれば整数が出ます。その整数が文書に書かれた価格かどうかは誰も保証しません。むしろ状況が微妙に悪くなることがあります。制約がないときモデルは「文書から価格を見つけられません」と形式を破って知らせることくらいはできるのに、制約がかかるとその逃げ道が塞がれ、どんな数字であれ出さなければなりません。形式のエラーが値のエラーに変わったわけで、値のエラーは検知がはるかに難しいのです。

ですから制約デコーディングを有効にするとき必ず一緒にやるべきことがあります。スキーマに「分からない」を表す場所を作ることです。

# 悪い: 値がなくても必ず数字を出さなければならない
{"price": {"type": "integer"}}

# 良い: 「なし」を表現できる
{"price": {"type": ["integer", "null"]},
 "price_found": {"type": "boolean"}}

第二の注意点は、文法がモデルの自然な生成経路と衝突するときです。制約は確率分布を切るものであってモデルの理解を変えるものではないので、モデルが行きたがっている経路がすべて塞がれると、残ったもののうち最も悪くないトークンが選ばれます。スキーマが複雑で不自然なほどこの現象は大きくなります。制約デコーディングを有効にしたら形式のエラーは消えたのに精度が下がったなら、スキーマを単純にするのが先です。

第三は性能です。ステップごとに許容トークンのマスクを計算する必要があるのでオーバーヘッドがあります。最近の実装は文法をあらかじめコンパイルしてキャッシュし、このコストのほとんどを相殺しますが、スキーマが毎リクエスト変わるとコンパイルのコストがそのまま表に出ます。スキーマの数が有限なら再利用されるように作ってください。

スキーマ設計がすなわちプロンプトです

モデルはスキーマを制約としてだけ受け取るのではなく、テキストとして読みます。ですからフィールド名と説明は指示文と同じ重みを持ちます。ここで得られる精度の改善が、プロンプトを磨いて得られるものより大きい場合が多くあります。

from pydantic import BaseModel, Field
from typing import Literal, Optional

# 悪いスキーマ
class BadTicket(BaseModel):
    type: str            # 何でも入ってくる
    date: str            # 形式がまちまち
    info: dict           # 構造がない

# 良いスキーマ
class Ticket(BaseModel):
    reasoning: str = Field(description="분류 근거를 두 문장 이내로")
    category: Literal["결제", "배송", "환불", "계정", "기타"]
    urgency: Literal["낮음", "보통", "높음"]
    order_id: Optional[str] = Field(None, description="주문번호. 본문에 없으면 null")
    due_date_iso: Optional[str] = Field(None, description="YYYY-MM-DD 형식")

四つが変わりました。

自由文字列を列挙型に変えました。これは精度とパースの安定性を同時に上げます。ダウンストリームで同じラベルの二通りの書き方を同一のものへ対応づけるコードも消えます。

フィールド名に形式を入れました。due_date_isodateよりはるかに強い指示です。名前だけで形式が伝われば、説明文に使うトークンが節約できます。

ないかもしれない値をnull許容にしました。必須文字列のままにするとモデルは空文字列やもっともらしい偽の注文番号を作り出します。

そして最も効果が大きい変更はreasoningを先頭に置いたことです。生成は左から右へ進むので、前に置かれたフィールドが後ろのフィールドの条件になります。根拠を先に書かせれば、分類結果はその根拠に条件づけられて生成されます。同じフィールドを最後に置くと、すでに決まった答えに対する事後説明になり、精度改善の効果が消えます。JSONオブジェクトのキーの順序は意味上無関係ですが、生成の順序はまったく無関係ではありません

もう一つ。ネストは浅く保ってください。3段以上ネストしたオブジェクトは形式エラー率と値エラー率がともに上がり、制約デコーディングの文法も大きくなります。ネストが必要に見えるときは、たいてい呼び出しを二回に分けるほうが良いです。

検証・再試行ループをきちんと実装する

防御をすべて積んでも失敗は残ります。残った失敗の処理の仕方が成功率を大きく分けます。

肝心なのは再試行するときに何を変えるかです。同じプロンプトでただもう一度呼ぶのは温度が0でないときにだけ意味があり、それでも同じ失敗を繰り返す確率が高いのです。エラーメッセージを会話に返してやると成功率がはっきり変わります。

import json
from pydantic import ValidationError

def generate_structured(client, messages, model_cls, max_attempts=3, max_tokens=1024):
    history = list(messages)
    last_error = None

    for attempt in range(max_attempts):
        resp = client.create(messages=history, max_tokens=max_tokens)
        text = resp.content
        finish = resp.finish_reason

        if finish == "length":
            # 途中で切れたのは再生成では解けない。上限を上げるか作業を分割する。
            max_tokens = min(max_tokens * 2, 8192)
            history.append({"role": "user",
                            "content": "출력이 잘렸습니다. 더 짧게, 설명 없이 다시 출력하십시오."})
            continue

        try:
            return model_cls.model_validate(extract_json(text))
        except (ValueError, json.JSONDecodeError, ValidationError) as e:
            last_error = e
            # 失敗した出力と具体的なエラーを一緒に返してやる
            history.append({"role": "assistant", "content": text})
            history.append({"role": "user", "content":
                f"위 출력이 스키마 검증에 실패했습니다.\n오류: {e}\n"
                f"오류만 수정한 JSON을 다른 텍스트 없이 출력하십시오."})

    raise RuntimeError(f"{max_attempts}회 시도 후 실패: {last_error}")

三つを指摘します。

エラーメッセージは具体的でなければなりません。「検証失敗」ではなく「urgencyフィールドに与えられた値は許可された値ではありません」まで伝えて初めてモデルは直せます。pydanticやjsonschemaのエラー文字列をそのまま渡せばたいてい十分です。

失敗した出力そのものを会話に入れる必要があります。モデルは自分が何を書いたかを見て初めて修正できます。

途中で切れたのは別の失敗です。同じやり方で再試行すれば同じ地点でまた切れます。終了理由を確認して分岐する必要があります。

再試行のコストも計算しておくと良いです。

def retry_math(p_fail, max_attempts):
    expected_calls = sum(p_fail ** i for i in range(max_attempts))
    residual = p_fail ** max_attempts
    return expected_calls, residual

for p in (0.04, 0.15):
    calls, res = retry_math(p, 3)
    print(f"1회 실패율 {p:.0%} → 평균 호출 {calls:.3f}회 (비용 +{calls - 1:.1%}), 최종 실패율 {res:.4%}")
# 1회 실패율 4% → 평균 호출 1.042회 (비용 +4.2%), 최종 실패율 0.0064%
# 1회 실패율 15% → 평균 호출 1.172회 (비용 +17.2%), 최종 실패율 0.3375%

ここで独立試行を仮定している点は指摘しておく必要があります。実際の失敗は相関しています。特定の入力が難しくて失敗したのなら、二回目の試行も同じ理由で失敗する確率が高いのです。ですから上の最終失敗率は楽観的な推定値であり、実際の値はログで測る必要があります。再試行の回数を3回以上に増やすのがほとんど役に立たない理由も、この相関のためです。

出力上限と途中切れへの備え

途中で切れるのは静かな失敗です。パースエラーとして表に出ればまだ幸運で、配列の最後の項目だけが消えたまま有効なJSONになる場合が最悪です。検証も通りアラートも鳴らないのにデータが欠けています。

def estimate_output_tokens(schema_fields, avg_value_tokens, list_len=1):
    """構造のオーバーヘッド + 値。実際の値はトークナイザで測るほうが正確。"""
    per_item = sum(len(name) // 3 + 4 + avg_value_tokens for name in schema_fields)
    return int(per_item * list_len * 1.25)   # 25%の余裕

fields = ["reasoning", "category", "urgency", "order_id", "due_date_iso"]
print(estimate_output_tokens(fields, avg_value_tokens=12, list_len=1))   # 116
print(estimate_output_tokens(fields, avg_value_tokens=12, list_len=20))  # 2325

一覧を抽出する作業で項目数が予測不能なら、上限の算定は根本的に不可能です。このとき使える方法が三つあります。

スキーマに項目数の上限を明示し、超過分を次の呼び出しに送るページネーションを作ります。あるいはJSON配列の代わりに一行に一オブジェクトずつ出力する方式に変え、切れても最後の行だけ捨てれば済むようにします。最後に、配列の長さを数えるフィールドを前のほうに置き、実際の配列の長さと比較して途中切れを事後検証します。

どちらにせよ終了理由の確認は必須です。この値を確認しないパイプラインは途中切れを絶対に検知できません。

おわりに — 形式は強制し、意味は検証してください

まとめると一文です。形式は生成段階で強制し、意味は生成の後に検証してください。

形式の強制はツールスキーマと制約デコーディングで原理的に解決します。ここに長く時間を使う理由はありません。一方、値が正しいか、ない値を作り出していないか、列挙値が実際の状況に対応しているかは、スキーマが絶対に捕まえてくれません。制約デコーディングを導入した後にパース失敗率が0になる瞬間が危険な理由がここにあります。ダッシュボードの赤いランプが消えただけで、間違った値はそのまま流れていきます。

ですからパース成功率の隣に値の正確度の指標も一緒に置いてください。そしてスキーマに手を入れるときは、フィールド名と順序から見てください。プロンプトを十回直すより、フィールドを一つ前に動かすほうが良い場合がよくあります。