8장: 출력 검증과 폴백 전략

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Self
from datetime import date
 
 
class ExtractedInvoice(BaseModel):
    """송장 추출 결과 (검증 포함)"""
    invoice_number: str = Field(
        min_length=1,
        max_length=50,
        description="송장 번호"
    )
    issue_date: str = Field(description="발행일 (YYYY-MM-DD)")
    due_date: str | None = Field(description="만기일 (YYYY-MM-DD)")
    vendor_name: str = Field(min_length=1, description="공급자 상호")
    buyer_name: str = Field(min_length=1, description="구매자 상호")
    subtotal: float = Field(ge=0, description="소계")
    tax_rate: float = Field(ge=0, le=1, description="세율 (0-1)")
    tax_amount: float = Field(ge=0, description="세액")
    total_amount: float = Field(ge=0, description="합계")
    currency: str = Field(description="통화 코드 (ISO 4217)")
 
    @field_validator("issue_date", "due_date")
    @classmethod
    def validate_date_format(cls, v: str | None) -> str | None:
        if v is None:
            return v
        try:
            date.fromisoformat(v)
        except ValueError:
            raise ValueError(f"날짜 형식이 올바르지 않습니다: {v} (YYYY-MM-DD 필요)")
        return v
 
    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        valid = {"KRW", "USD", "EUR", "JPY", "GBP", "CNY"}
        if v.upper() not in valid:
            raise ValueError(f"지원하지 않는 통화: {v}")
        return v.upper()
 
    @model_validator(mode="after")
    def validate_amounts(self) -> Self:
        """금액 정합성 검증"""
        expected_tax = round(self.subtotal * self.tax_rate, 2)
        if abs(expected_tax - self.tax_amount) > 1.0:
            raise ValueError(
                f"세액 불일치: 소계({self.subtotal}) x 세율({self.tax_rate}) = "
                f"{expected_tax}, 실제 세액: {self.tax_amount}"
            )
 
        expected_total = self.subtotal + self.tax_amount
        if abs(expected_total - self.total_amount) > 1.0:
            raise ValueError(
                f"합계 불일치: 소계({self.subtotal}) + 세액({self.tax_amount}) = "
                f"{expected_total}, 실제 합계: {self.total_amount}"
            )
        return self
 
    @model_validator(mode="after")
    def validate_dates(self) -> Self:
        """날짜 논리 검증"""
        if self.due_date and self.issue_date:
            if self.due_date < self.issue_date:
                raise ValueError(
                    f"만기일({self.due_date})이 "
                    f"발행일({self.issue_date})보다 앞섭니다."
                )
        return self

from pydantic import BaseModel, Field, model_validator
from typing import Literal, Self
 
 
class OrderData(BaseModel):
    """주문 데이터"""
    order_type: Literal["domestic", "international"]
    shipping_address: str
    country_code: str | None = Field(default=None)
    customs_number: str | None = Field(default=None)
    total_amount: float = Field(ge=0)
    currency: str
 
    @model_validator(mode="after")
    def validate_international_fields(self) -> Self:
        """해외 주문이면 국가 코드와 통관 번호가 필수"""
        if self.order_type == "international":
            if not self.country_code:
                raise ValueError("해외 주문에는 국가 코드가 필수입니다.")
            if not self.customs_number:
                raise ValueError("해외 주문에는 통관 번호가 필수입니다.")
        return self

from dataclasses import dataclass
 
 
@dataclass
class ValidationIssue:
    """검증 이슈"""
    field: str
    severity: str  # error, warning, info
    message: str
 
 
class BusinessValidator:
    """비즈니스 규칙 검증기"""
 
    def validate_invoice(self, invoice: ExtractedInvoice) -> list[ValidationIssue]:
        issues = []
 
        # 금액 범위 검증
        if invoice.total_amount > 100_000_000:
            issues.append(ValidationIssue(
                field="total_amount",
                severity="warning",
                message=f"합계가 1억 원을 초과합니다: {invoice.total_amount}"
            ))
 
        # 미래 날짜 검증
        from datetime import date
        today = date.today().isoformat()
        if invoice.issue_date > today:
            issues.append(ValidationIssue(
                field="issue_date",
                severity="error",
                message=f"발행일이 미래 날짜입니다: {invoice.issue_date}"
            ))
 
        # 사업자 등록 여부 확인 (외부 API 호출)
        if not self._verify_vendor(invoice.vendor_name):
            issues.append(ValidationIssue(
                field="vendor_name",
                severity="warning",
                message=f"등록되지 않은 공급자: {invoice.vendor_name}"
            ))
 
        return issues
 
    def _verify_vendor(self, vendor_name: str) -> bool:
        """공급자 등록 여부를 확인합니다 (예시)."""
        # 실제로는 데이터베이스나 외부 API를 조회
        registered_vendors = {"삼성전자", "LG전자", "현대모비스"}
        return vendor_name in registered_vendors

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
 
 
class SemanticValidation(BaseModel):
    """의미적 검증 결과"""
    is_faithful: bool = Field(
        description="추출 결과가 원본에 충실한지 여부"
    )
    hallucinated_fields: list[str] = Field(
        default_factory=list,
        description="원본에 없는 내용이 생성된 필드 목록"
    )
    missing_fields: list[str] = Field(
        default_factory=list,
        description="원본에 있으나 추출되지 않은 정보"
    )
    accuracy_score: float = Field(
        ge=0.0, le=1.0,
        description="추출 정확도 점수 (0-1)"
    )
    details: str = Field(description="검증 상세 설명")
 
 
def validate_semantically(
    original_text: str,
    extracted_data: dict,
    schema_description: str
) -> SemanticValidation:
    """추출 결과의 의미적 정확성을 검증합니다."""
    client = instructor.from_openai(OpenAI())
 
    return client.chat.completions.create(
        model="gpt-4o-mini",  # 검증에는 경량 모델 사용
        response_model=SemanticValidation,
        messages=[
            {
                "role": "system",
                "content": (
                    "당신은 데이터 추출 결과를 검증하는 전문가입니다. "
                    "원본 텍스트와 추출된 데이터를 비교하여 "
                    "정확성을 평가하세요. "
                    "특히 다음을 확인합니다:\n"
                    "1. 환각(hallucination): 원본에 없는 내용이 만들어졌는가\n"
                    "2. 누락: 원본에 있는 중요 정보가 빠졌는가\n"
                    "3. 변조: 값이 잘못 추출되었는가"
                )
            },
            {
                "role": "user",
                "content": (
                    f"스키마: {schema_description}\n\n"
                    f"원본 텍스트:\n{original_text}\n\n"
                    f"추출 결과:\n{extracted_data}"
                )
            }
        ]
    )

import instructor
from openai import OpenAI
from pydantic import BaseModel, ValidationError
from tenacity import retry, stop_after_attempt, wait_exponential
import json
 
 
class ExtractionResult(BaseModel):
    """추출 결과"""
    data: dict
    confidence: float
    warnings: list[str]
 
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def extract_with_custom_retry(
    content: str,
    response_model: type[BaseModel],
    validation_fn=None
) -> BaseModel:
    """커스텀 검증 함수를 포함한 추출"""
    client = instructor.from_openai(OpenAI())
 
    result = client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=response_model,
        messages=[
            {"role": "system", "content": "데이터를 정확히 추출하세요."},
            {"role": "user", "content": content}
        ]
    )
 
    # 추가 커스텀 검증
    if validation_fn:
        issues = validation_fn(result)
        if issues:
            raise ValueError(
                f"커스텀 검증 실패: {'; '.join(issues)}"
            )
 
    return result

from pydantic import BaseModel, ValidationError
 
 
def extract_with_enriched_context(
    content: str,
    response_model: type[BaseModel],
    max_attempts: int = 3
) -> BaseModel:
    """검증 오류 컨텍스트를 축적하며 재시도합니다."""
    import instructor
    from openai import OpenAI
 
    client = instructor.from_openai(OpenAI())
    error_history = []
 
    for attempt in range(1, max_attempts + 1):
        # 이전 오류를 시스템 메시지에 포함
        system_content = "데이터를 정확히 추출하세요."
        if error_history:
            error_summary = "\n".join(
                f"- 시도 {i+1}: {err}"
                for i, err in enumerate(error_history)
            )
            system_content += (
                f"\n\n이전 시도에서 다음 오류가 발생했습니다. "
                f"이 오류를 반드시 수정하세요:\n{error_summary}"
            )
 
        try:
            result = client.chat.completions.create(
                model="gpt-4o-2026-02",
                response_model=response_model,
                messages=[
                    {"role": "system", "content": system_content},
                    {"role": "user", "content": content}
                ]
            )
            return result
 
        except (ValidationError, ValueError) as e:
            error_history.append(str(e))
            if attempt == max_attempts:
                raise
 
    raise RuntimeError("최대 재시도 횟수 초과")

import instructor
from openai import OpenAI
from anthropic import Anthropic
from pydantic import BaseModel
from typing import TypeVar
import logging
 
T = TypeVar("T", bound=BaseModel)
logger = logging.getLogger(__name__)
 
 
class ProviderConfig:
    """프로바이더 설정"""
    def __init__(self, name: str, client, model: str, priority: int):
        self.name = name
        self.client = client
        self.model = model
        self.priority = priority
 
 
class FallbackChain:
    """멀티 프로바이더 폴백 체인"""
 
    def __init__(self):
        self.providers: list[ProviderConfig] = []
 
    def add_provider(self, config: ProviderConfig) -> "FallbackChain":
        self.providers.append(config)
        self.providers.sort(key=lambda p: p.priority)
        return self
 
    def extract(
        self,
        response_model: type[T],
        messages: list[dict],
        max_retries: int = 2
    ) -> T:
        """폴백 체인을 통해 추출을 시도합니다."""
        errors = []
 
        for provider in self.providers:
            try:
                logger.info(f"프로바이더 시도: {provider.name}")
 
                if provider.name == "openai":
                    result = provider.client.chat.completions.create(
                        model=provider.model,
                        response_model=response_model,
                        max_retries=max_retries,
                        messages=messages,
                    )
                elif provider.name == "anthropic":
                    result = provider.client.messages.create(
                        model=provider.model,
                        max_tokens=4096,
                        response_model=response_model,
                        max_retries=max_retries,
                        messages=messages,
                    )
                else:
                    raise ValueError(f"지원하지 않는 프로바이더: {provider.name}")
 
                logger.info(f"성공: {provider.name}")
                return result
 
            except Exception as e:
                logger.warning(f"실패: {provider.name} - {e}")
                errors.append((provider.name, str(e)))
 
        # 모든 프로바이더 실패
        error_details = "; ".join(
            f"[{name}] {err}" for name, err in errors
        )
        raise RuntimeError(f"모든 프로바이더 실패: {error_details}")
 
 
# 사용 예시
chain = FallbackChain()
chain.add_provider(ProviderConfig(
    name="openai",
    client=instructor.from_openai(OpenAI()),
    model="gpt-4o-2026-02",
    priority=1,
))
chain.add_provider(ProviderConfig(
    name="anthropic",
    client=instructor.from_anthropic(Anthropic()),
    model="claude-sonnet-4-20250514",
    priority=2,
))

import json
from pydantic import BaseModel, Field, ValidationError
 
 
class PartialInvoice(BaseModel):
    """부분 복구가 가능한 송장 데이터"""
    invoice_number: str | None = None
    vendor_name: str | None = None
    total_amount: float | None = None
    date: str | None = None
    items: list[dict] = Field(default_factory=list)
    extraction_completeness: float = Field(
        default=0.0,
        description="추출 완성도 (0-1)"
    )
 
 
def recover_partial_output(raw_output: str) -> PartialInvoice:
    """불완전한 JSON 출력에서 가능한 데이터를 복구합니다."""
 
    # 1. 정상 JSON 파싱 시도
    try:
        data = json.loads(raw_output)
        return PartialInvoice(**data)
    except (json.JSONDecodeError, ValidationError):
        pass
 
    # 2. 불완전한 JSON 복구 시도
    try:
        # 닫히지 않은 괄호 보완
        fixed = raw_output.rstrip()
        open_braces = fixed.count("{") - fixed.count("}")
        open_brackets = fixed.count("[") - fixed.count("]")
        fixed += "]" * open_brackets + "}" * open_braces
 
        data = json.loads(fixed)
        result = PartialInvoice(**data)
 
        # 채워진 필드 수로 완성도 계산
        total_fields = 5  # 핵심 필드 수
        filled = sum(1 for v in [
            result.invoice_number,
            result.vendor_name,
            result.total_amount,
            result.date,
            result.items
        ] if v)
        result.extraction_completeness = filled / total_fields
        return result
 
    except (json.JSONDecodeError, ValidationError):
        pass
 
    # 3. 정규표현식으로 개별 필드 추출
    import re
    result = PartialInvoice()
 
    number_match = re.search(r'"invoice_number"\s*:\s*"([^"]+)"', raw_output)
    if number_match:
        result.invoice_number = number_match.group(1)
 
    amount_match = re.search(r'"total_amount"\s*:\s*([\d.]+)', raw_output)
    if amount_match:
        result.total_amount = float(amount_match.group(1))
 
    return result

from dataclasses import dataclass
from enum import Enum
 
 
class ValidationStatus(str, Enum):
    PASSED = "passed"
    WARNING = "warning"
    FAILED = "failed"
 
 
@dataclass
class ValidationReport:
    """검증 보고서"""
    status: ValidationStatus
    schema_valid: bool
    business_valid: bool
    semantic_valid: bool
    issues: list[str]
    confidence: float
 
 
async def run_validation_pipeline(
    original_text: str,
    extracted_data: BaseModel,
    business_validator: BusinessValidator
) -> ValidationReport:
    """다중 레이어 검증 파이프라인을 실행합니다."""
    issues = []
 
    # 1. 스키마 검증 (Pydantic이 이미 처리)
    schema_valid = True
 
    # 2. 비즈니스 로직 검증
    biz_issues = business_validator.validate_invoice(extracted_data)
    biz_errors = [i for i in biz_issues if i.severity == "error"]
    business_valid = len(biz_errors) == 0
    issues.extend(i.message for i in biz_issues)
 
    # 3. 의미적 검증
    semantic_result = validate_semantically(
        original_text=original_text,
        extracted_data=extracted_data.model_dump(),
        schema_description="송장 데이터"
    )
    semantic_valid = semantic_result.is_faithful
    if not semantic_valid:
        issues.extend(
            f"환각 필드: {f}" for f in semantic_result.hallucinated_fields
        )
        issues.extend(
            f"누락 필드: {f}" for f in semantic_result.missing_fields
        )
 
    # 종합 판정
    if not schema_valid or not business_valid:
        status = ValidationStatus.FAILED
    elif not semantic_valid:
        status = ValidationStatus.WARNING
    else:
        status = ValidationStatus.PASSED
 
    return ValidationReport(
        status=status,
        schema_valid=schema_valid,
        business_valid=business_valid,
        semantic_valid=semantic_valid,
        issues=issues,
        confidence=semantic_result.accuracy_score
    )