让 LLM 输出不再失控:用 Pydantic 校验大模型输出 – wang_yb

让 LLM 输出不再失控:用 Pydantic 校验大模型输出 – wang_yb


あなたは prompt に書かれています 「JSONを返してください」モデルによって返される高い確率はクリーンな期間ではありません JSONただし段落です。最初は「OK、これが抽出結果です」、真ん中は “`json フェンス、最後は「お役に立てば幸いです」です。

フィールド自体は信頼できない可能性があります。たとえば、勤務年数は「5 年」、期待給与は「約 15,000」と書かれており、場合によっては 1 つまたは 2 つのキーが欠けていることがあります。

これらのエラーの中で最も負荷が高いのは、エラーを報告するエラーではなく、エラーを報告しないエラーです。 "15k" これはデータベースに数値として書き込まれます。 3か月後にレポートに異常が見つかった場合、どのデータが混入していたのか分からなくなります。

「履歴書の本文から構造化情報を抽出する」を例に考えてみましょう。検証なしと検証ありの 2 つのメソッドを実行して違いを確認し、直接転送できる一連の記述メソッド (JSON の抽出、スキーマの定義、型と範囲の検証、クロスフィールド検証、失敗後の再試行) を示します。

シミュレーション データはプロセス全体で使用され、API キーは必要ありません。pip install pydantic[email] 走れる。

最も単純な記述方法を使用して同じモデルが返されると、次の 4 つの状況が連続して発生します。

json.loads(llm_text)              # ① 外层裹了话
data["years_experience"] + 1      # ② 值是 "五年"
len(data["skills"])               # ③ 压根没这个 key
"15k左右" * 12                    # ④ 算年薪

実際の出力:

JSONDecodeError: Expecting value: line 2 column 1 (char 1)
TypeError: can only concatenate str (not "int") to str
KeyError: 'skills'
'15k左右' * 12 = 15k左右15k左右15k左右…(重复 12 次)

①②③ 幸運なことに、プログラムはその場で爆発します。せいぜいエラーが発生した場所から少し離れた場所です——KeyError 多くの場合、ダウンストリームでレポートを生成する行が、このコードの解析からいくつかの関数によって分離されて表示されます。

本当の問題は④: 文字列の乗算は Python では完全に合法な演算です。例外をスローしたり、ログを書き込んだりしません。その結果、誰も見たことがないものが次々と生まれ、それがあなたの年収欄に入ります。

共通点は次のとおりです。問題の発見が遅すぎるか、まったく発見されないかのどちらかです。

json.loads 「これは合法です」と保証してください JSON 文法』では、「フィールドがあるかどうか、型が正しいかどうか、値が妥当かどうか」については何も書かれていません。

検証が追加されると、リンクは「テキストがビジネスに直接入る」から 2 つのレベルに変わります。まず、単語の束から JSON を抽出し、次にスキーマに従ってフィールドごとに検証します。

让 LLM 输出不再失控:用 Pydantic 校验大模型输出 – wang_yb

最初のレベルは非常に単純です。マークダウン フェンスを削除し、正規表現を使用して外側の中括弧のペアをキャプチャします。

def extract_json_from_llm_response(response: str) -> dict:
    text = re.sub(r"```(?:json)?", "", response)      # 去掉 ``` 围栏
    match = re.search(r"\{.*\}", text, re.DOTALL)     # 取最外层花括号
    if not match:
        raise ValueError("响应里没找到 JSON")
    return json.loads(match.group())

2 番目のパスは次のパスに渡されます。 Pydantic。

2 つのレイヤーを分離する価値はエラー メッセージにあります —JSONDecodeError 「モデルは形式に従って話さない」と説明し、ValidationError 「形式は正しいが、内容が間違っている」ことを説明します。 2 つの状況はまったく異なる方法で処理されるため (この違いはセクション 5 で後で使用します)、これらを同じものにしないでください。 except Exception。

を定義する Resumeフィールドと制約は一緒に記述されます。

class Resume(BaseModel):
    name: str
    email: EmailStr
    phone: str
    city: str
    expected_salary: int = Field(..., gt=0, le=100_000)
    skills: List[str] = Field(..., min_length=1)
    experiences: List[WorkExperience] = Field(..., min_length=1)
    years_experience: float = Field(..., ge=0, le=40)

EmailStr 電子メール形式を直接確認します (インストールが必要です) email-validatorつまり pydantic[email] その余分な);

Field(gt=0, le=100_000) 値の範囲をタイプヒントに書き込みます。

フィールドはデフォルトで必須です。記入漏れがある場合は、その場で報告してください。 Field requiredプラス min_length=1 リストが空でないことを確認してください。

これらの行はセクション 1 の③をブロックします。欠落しているフィールドは入り口で失敗し、下流までスローされません。 KeyError。

興味深いのはカスタムバリデーターです。値を確実に変換できる場合は、エラーを報告させずに、簡単に修正してください。:

@field_validator("phone")
@classmethod
def normalize_phone(cls, v: str) -> str:
    digits = re.sub(r"\D", "", v)                      # 去掉 "-"、空格
    if digits.startswith("86") and len(digits) == 13:  # 去掉 +86 前缀
        digits = digits[2:]
    if len(digits) != 11 or not digits.startswith("1"):
        raise ValueError(f"不是有效的手机号:{v!r}")
    return digits

@field_validator("expected_salary", mode="before")
@classmethod
def parse_salary(cls, v):
    """把 '15k' / '1.5万' 先转成数字,再交给 int 校验。"""
    ...

mode="before" は「型チェックの前に実行してください」という意味なので、元の文字列が取得されます。

もう一度実行して効果を確認します。

手机号 '138-0013-8000' -> 13800138000
薪资   '15k左右'       -> 15000

乱雑な形式だがセマンティクスは明確な入力が救済され、ライブラリには常にきちんとした 11 桁の携帯電話番号と整数値が保存されます。


保存できない人のみエラーが報告されます。「5 年」など – モデルが 5 年なのか 5.5 年なのかを推測することはできませんし、推測できないので、失敗させます。

Input should be a valid number, unable to parse string as a number [input_value="五年"]

フィールド名、理由、および元の値はすべてこのメッセージに含まれており、モデルに直接フィードバックできます (以下のセクション 5)。

判断基準はただ一つ。変換ルールは独特ですか?——「はい」の場合は基準値に換算します。そうでない場合は、エラーが報告されます。

実際のデータが横ばいになることはほとんどありません。

履歴書に職歴のリストがある場合は、サブモデルを定義してメイン モデルに書き込みます。 List[WorkExperience]:

Pydantic サブオブジェクトは再帰的に構築されます。各項目のどのフィールドにエラーがある場合、エラー メッセージは「2 番目に発生した項目の開始」まで正確になります。

ただし、範囲と形式の検証は「形式エラー」を防ぐことしかできず、モデルによる深刻なデータのコンパイルを防ぐことはできません。たとえば、3 つの経験期間を合計すると 7 年になりますが、合計すると 9 年になります。

現時点では、フィールド全体をチェックする必要があります。

@field_validator("years_experience")
@classmethod
def check_years_match(cls, v: float, info) -> float:
    experiences = info.data.get("experiences") or []
    if experiences:
        total = sum(_years_of(e.start, e.end) for e in experiences)
        if abs(total - v) > 0.5:                       # 留 0.5 年的容差
            raise ValueError(f"总工作年限 {v} 年与各段经历合计 {total:.2f} 年对不上")
    return v

足りなくなったのはこんな感じです。

Value error, 总工作年限 9.0 年与各段经历合计 7.00 年对不上

この種の「自分と一致しない」という問題は、LLM 出力で注意すべき最も重要なカテゴリです。各フィールド単独では正当ですが、すべてを組み合わせると偽になります。

同じ考えが多くの場所で使用できます。注文詳細の合計は注文総額と等しくなければならず、日次レポートの顧客価格は売上高を注文数で割ったものと等しくなければならず、抽出された日付範囲は元に戻すことはできません。

次のように書くときに落とし穴があります。バリデーターは、以前に検証したフィールドのみを読み取ることができます、それで years_experience で宣言する必要があります experiences 後ろ、そうでない場合 info.data 空いてますよ。

さらに、許容差が 0 に設定されている場合、モデルによって計算された合計は数十分の 1 つ異なることがよくあります。

検証が失敗した場合はどうすればよいですか?最も簡単で効果的な方法は、最後のエラー メッセージをプロンプトに入力し、再度応答させます。

def extract_with_retry(llm_call_function, max_retries: int = 3):
    last_error = None
    for attempt in range(max_retries):
        try:
            return Resume(**extract_json_from_llm_response(llm_call_function(last_error)))
        except ValidationError as e:
            last_error = str(e)                        # schema 错:带上具体字段问题
        except json.JSONDecodeError:
            last_error = "返回不是合法 JSON,请只输出 JSON,不要任何解释文字。"
        if attempt == max_retries - 1:
            return None                                # 用尽次数,交给上层降级

llm_call_function(last_error) スペルミスの責任を負う promptしたがって、モデルはどこを変更すべきかを認識します。

同じ履歴書テキストを 3 回実行したときの実際の出力は次のとおりです。

第 1 次失败:Input should be a valid number …("五年")
第 2 次失败:总工作年限 9.0 年与各段经历合计 7.00 年对不上
成功:陈舒雅,7.0 年经验,期望 15000 元,技能 ['Python', 'SQL', 'FineBI']

3 つの詳細に注意する必要があります。

  1. JSON エラーとスキーマ エラーでは異なるプロンプトが使用されます前者は「フォーマットに従っていない」ことを意味し、後者は「特定のフィールドが間違っている」ことを意味します。
  2. 上限があるはずだ回数を使いきったら返却 None プロセス全体を崩壊させるのではなく、上位層をダウングレードさせます (手動レビューキューを書き込み、アラームをログに記録します)。
  3. 電話をかけるとき temperature 0 に設定しないと、同じプロンプトに対して毎回エラーが異なり、再試行は無駄になります。

前の例では、シミュレートされたリターンを使用しました。これを実際の API に置き換えるには、あと 2 つの手順だけが必要です。 schema ぬいぐるみ prompt戻り値を同じものに渡します parse_resume。

schema_hint = json.dumps(Resume.model_json_schema(), ensure_ascii=False)
response = client.chat.completions.create(
    model="gpt-4o-mini", temperature=0,
    messages=[{"role": "user", "content": f"按这个 JSON Schema 抽取:\n{schema_hint}\n\n{text}"}])
return parse_resume(response.choices[0].message.content)

フレームワークを使用すると簡単です。LangChain 持っている llm.with_structured_output(Resume)(ネイティブ関数呼び出しを使用することをお勧めします) PydanticOutputParser(プロンプトに従ってガイドされ、任意のモデルを使用できます)。

LlamaIndex には LLMTextCompletionProgram.from_defaults(output_cls=Resume)。

しかしフレームワークが制約するのに役立つのは「ビジネスの能力」ではなく「構造」です。function calling フィールドとタイプが正しいことは保証できますが、「合計年数は各経験期間の合計に等しい」などのビジネスのルールを保証することはできません。

さらに、構造化された出力は、極端な場合には依然として非準拠となります。

したがって、Pydantic 検証の最後の層は保持することが推奨されます。また、任意のモデル サプライヤーに接続する場合の統合アウトレットでもあり、モデルを変更するときにダウンストリームを変更する必要はありません。

完全な実行可能なスクリプトは次の場所で共有されています。resume_extraction_demo.py

内部の「モデルの戻り値」はすべてシミュレートされた文字列であり、インストールされています。 pydantic[email] 直接 python resume_extraction_demo.py この記事では、各出力セットを確認できます。

シミュレートされたコードの実行結果は次のとおりです。

独自のシーンを置き換えるのは簡単です。

まずは機種変更です。 バンドル Resume フィールドを自分のものに置き換えます: 書類の日付、金額、カテゴリ、ID、担当者。まず「must have」と「may have」を区別し、後者を書きます。 Optional[str] = Noneすべてを必要に応じて設定しないでください。そうしないと、モデルは毎回ノックバックされます。数値フィールドは常に追加されます gt / ge / leこのタイプの制約は最も安価で効果的です。

2 番目のステップは、分析を変更することです。 extract_json_from_llm_response 階層化された例外を直接コピーし、プロンプトに追加します。 model_json_schema() の出力temperature=0。

3番目のステップはポケットを変更することです。 失敗→2~3回リトライ→それでもダメならリターン None 手動で行うか、ダウングレードします。 1 つの解析失敗によってバッチ処理タスク全体がダウンしないようにしてください。

最初に戻る 「JSONを返す」 献身。モデルがエラーを起こすかどうかは制御できません。ただし、「エラー発生後にその場で停止するか、静かにデータベースに流し込むか」は完全にあなたの手の中にあります。

違いは、コントラクトがコードに記述されているかどうかです。書き留めておくと、一文だけではなく、呼び出されるたびに実行されるチェックになります。 prompt 口頭合意は有効です。



Source link

Leave a Reply

Your email address will not be published. Required fields are marked *