不正な出力の検証と再試行

LLMの出力に検証が必要な理由

検証のレイヤー

形式チェックのためのPydanticバリデーター

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import re
from datetime import datetime

class ExtractedInvoice(BaseModel):
    vendor: str
    invoice_number: Optional[str]
    amount: float = Field(gt=0, description='Must be positive')
    currency: str = Field(min_length=3, max_length=3)
    invoice_date: str

    @field_validator('currency')
    @classmethod
    def currency_must_be_uppercase(cls, v):
        return v.upper()

    @field_validator('invoice_date')
    @classmethod
    def parse_date(cls, v):
        # Try to parse common date formats
        for fmt in ('%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y', '%B %d, %Y'):
            try:
                datetime.strptime(v, fmt)
                return v
            except ValueError:
                continue
        raise ValueError(f'Cannot parse date: {v}')

    @field_validator('amount')
    @classmethod
    def reasonable_amount(cls, v):
        if v > 10_000_000:
            raise ValueError(f'Amount {v} seems unreasonably large. Flag for review.')
        return round(v, 2)

修正フィードバック付き再試行パターン

import openai
from pydantic import BaseModel, ValidationError, Field

client = openai.OpenAI()

class PriceExtraction(BaseModel):
    product: str
    price_usd: float = Field(gt=0, lt=100000)
    quantity: int = Field(ge=1)

def extract_with_retry(text: str, max_retries: int = 3) -> PriceExtraction:
    messages = [
        {'role': 'system', 'content': 'Extract product pricing information.'},
        {'role': 'user', 'content': text}
    ]

    for attempt in range(max_retries):
        result = client.beta.chat.completions.parse(
            model='gpt-4o-mini',
            messages=messages,
            response_format=PriceExtraction
        )
        msg = result.choices[0].message
        if msg.refusal:
            raise ValueError(f'Model refused: {msg.refusal}')

        try:
            return msg.parsed  # Pydantic validates on parse
        except ValidationError as e:
            if attempt == max_retries - 1:
                raise
            # Add corrective feedback for the next attempt
            messages.append({'role': 'assistant', 'content': msg.content})
            messages.append({'role': 'user', 'content': f'The previous extraction failed validation: {e}\nPlease correct and try again.'})
            print(f'Attempt {attempt+1} failed. Retrying with feedback...')

ビジネスロジック検証

from pydantic import BaseModel, Field, model_validator
from typing import List

class LineItem(BaseModel):
    description: str
    quantity: int = Field(ge=1)
    unit_price: float = Field(ge=0)
    line_total: float

    @model_validator(mode='after')
    def check_line_total(self):
        expected = round(self.quantity * self.unit_price, 2)
        actual = round(self.line_total, 2)
        if abs(expected - actual) > 0.02:  # Allow 2-cent rounding tolerance
            raise ValueError(
                f'Line total {actual} does not match quantity*price={expected}'
            )
        return self

class Invoice(BaseModel):
    line_items: List[LineItem]
    subtotal: float
    tax: float
    total: float

    @model_validator(mode='after')
    def check_invoice_total(self):
        expected_total = round(self.subtotal + self.tax, 2)
        if abs(expected_total - round(self.total, 2)) > 0.02:
            raise ValueError(
                f'Invoice total {self.total} != subtotal+tax ({expected_total})'
            )
        return self

検証失敗のログ記録

import logging
from pydantic import ValidationError

logger = logging.getLogger(__name__)

def extract_with_logging(text: str, doc_id: str) -> dict:
    result = None
    for attempt in range(3):
        try:
            result = run_extraction(text)  # Your extraction function
            logger.info('Extraction success', extra={
                'doc_id': doc_id,
                'attempt': attempt + 1
            })
            return result
        except ValidationError as e:
            logger.warning('Validation failure', extra={
                'doc_id': doc_id,
                'attempt': attempt + 1,
                'errors': e.errors(),
                'error_count': len(e.errors())
            })
    # All retries failed
    logger.error('Extraction failed after max retries', extra={'doc_id': doc_id})
    return {'error': 'extraction_failed', 'doc_id': doc_id}

def run_extraction(text):
    pass  # Placeholder for actual extraction logic

確信度スコアとしきい値

from pydantic import BaseModel, Field
from typing import Optional

class ExtractedWithConfidence(BaseModel):
    value: Optional[str]
    confidence: float = Field(ge=0.0, le=1.0)
    reason: Optional[str] = None  # Why confidence is low, if below threshold

class DocumentExtraction(BaseModel):
    vendor_name: ExtractedWithConfidence
    invoice_amount: ExtractedWithConfidence
    due_date: ExtractedWithConfidence

def route_by_confidence(extraction: DocumentExtraction, threshold=0.85):
    low_confidence_fields = []
    for field_name, field_val in extraction.model_dump().items():
        if isinstance(field_val, dict) and field_val.get('confidence', 1.0) < threshold:
            low_confidence_fields.append(field_name)

    if not low_confidence_fields:
        return 'auto_approve'
    elif len(low_confidence_fields) > 2:
        return 'human_review'
    else:
        return f'spot_check: {low_confidence_fields}'

再試行に失敗した場合のフォールバック戦略

外部データに対する意味検証

from typing import Optional

# Simulated external data source
KNOWN_VENDORS = {'acme corp', 'techsupplies inc', 'globex corporation'}

def validate_against_crm(extraction: dict) -> dict:
    vendor = extraction.get('vendor', '').lower()
    warnings = []

    if vendor and vendor not in KNOWN_VENDORS:
        warnings.append({
            'field': 'vendor',
            'issue': f'Vendor "{vendor}" not found in CRM',
            'severity': 'warning'
        })
        # Optionally suggest closest match
        # from difflib import get_close_matches
        # matches = get_close_matches(vendor, KNOWN_VENDORS, n=1, cutoff=0.8)
        # if matches: warnings[-1]['suggestion'] = matches[0]

    return {
        'extraction': extraction,
        'warnings': warnings,
        'requires_review': len(warnings) > 0
    }

print('Semantic validation pattern defined')

バリデーションパイプラインのテスト

from pydantic import ValidationError

def test_extraction_validation():
    # Test cases: (input_data, should_pass)
    test_cases = [
        ({'vendor': 'ACME', 'amount': 150.00, 'currency': 'USD', 'invoice_date': '2025-01-15'}, True),
        ({'vendor': 'ACME', 'amount': -50.00, 'currency': 'USD', 'invoice_date': '2025-01-15'}, False),  # negative
        ({'vendor': 'ACME', 'amount': 150.00, 'currency': 'EURO', 'invoice_date': '2025-01-15'}, False), # 4-char
        ({'vendor': 'ACME', 'amount': 150.00, 'currency': 'USD', 'invoice_date': 'yesterday'}, False),   # bad date
    ]
    passed = failed = 0
    for data, should_pass in test_cases:
        try:
            # ExtractedInvoice(**data)  # Your Pydantic model
            if should_pass:
                passed += 1
            else:
                print(f'MISSED: Should have failed for {data}')
                failed += 1
        except (ValidationError, ValueError):
            if not should_pass:
                passed += 1
            else:
                print(f'UNEXPECTED FAIL for {data}')
                failed += 1
    print(f'Tests: {passed} passed, {failed} failed')

test_extraction_validation()

パターン別の失敗分析とプロンプトの調整

クイックチェック

レッスンのまとめ