스키마 기반 데이터 추출

스키마 기반 추출이 필요한 이유

프롬프트에 스키마 제공

import anthropic, json

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

INVOICE_SCHEMA = '''
{
  "invoice_number": "string",
  "vendor_name": "string",
  "vendor_address": "string or null",
  "invoice_date": "YYYY-MM-DD",
  "due_date": "YYYY-MM-DD or null",
  "line_items": [
    {
      "description": "string",
      "quantity": "number",
      "unit_price": "number",
      "total": "number"
    }
  ],
  "subtotal": "number",
  "tax": "number or null",
  "total_amount": "number",
  "currency": "3-letter ISO code e.g. USD"
}
'''

def extract_invoice(invoice_text):
    prompt = f'Extract structured data from this invoice.\nReturn JSON matching this schema exactly:\n{INVOICE_SCHEMA}\n\nInvoice:\n{invoice_text}'
    r = client.messages.create(model='claude-opus-4-5', max_tokens=500, messages=[{'role': 'user', 'content': prompt}])
    return json.loads(r.content[0].text)

print('Invoice schema defined.')

청구서 추출 예시

invoice_text = '''
INVOICE #INV-2025-0342
From: Acme Software Ltd.
123 Tech Street, San Francisco, CA 94105

Date: March 15, 2025
Due: April 14, 2025

Items:
- Annual Pro License (5 seats) x1 @ $2,400.00 = $2,400.00
- Setup & Onboarding x2 @ $300.00 = $600.00

Subtotal: $3,000.00
Tax (8.5%): $255.00
TOTAL DUE: $3,255.00 USD
'''

result = extract_invoice(invoice_text)
print(f'Invoice: {result["invoice_number"]}')
print(f'Vendor: {result["vendor_name"]}')
print(f'Total: {result["currency"]} {result["total_amount"]}')
print(f'Line items: {len(result["line_items"])}')

회의 메모 추출

MEETING_SCHEMA = '''
{
  "meeting_title": "string",
  "date": "YYYY-MM-DD",
  "attendees": ["string"],
  "decisions": ["string"],
  "action_items": [
    {
      "task": "string",
      "owner": "string or null",
      "due_date": "YYYY-MM-DD or null"
    }
  ],
  "next_meeting": "string or null"
}
'''

meeting_notes = '''
Product Sync - March 20, 2025
Attendees: Sarah (PM), Jake (Engineering), Priya (Design)

Decided to push the v2.0 launch to April 15.
Will not include the analytics dashboard in v2.0.

Actions:
- Jake to fix the login bug by March 25
- Priya to finalize mockups by March 22
- Sarah to send updated roadmap to stakeholders (no date set)

Next sync: March 27, same time.
'''

print(f'Meeting schema: {len(MEETING_SCHEMA)} chars')
print(f'Notes length: {len(meeting_notes)} chars')

제품 사양 추출

PRODUCT_SCHEMA = '''
{
  "product_name": "string",
  "sku": "string or null",
  "category": "string",
  "price": {"amount": "number", "currency": "string"},
  "dimensions": {
    "length_cm": "number or null",
    "width_cm": "number or null",
    "height_cm": "number or null",
    "weight_kg": "number or null"
  },
  "colors": ["string"],
  "materials": ["string"],
  "features": ["string"],
  "in_stock": true | false
}
'''

product_text = 'AlphaDesk Pro standing desk. SKU: AD-PRO-001. $899. Available in white and black. 120x60x75cm, 35kg. Steel frame, bamboo top. Features: memory height, anti-collision, app control. In stock.'

prompt = f'Extract product specs. Return JSON:\n{PRODUCT_SCHEMA}\n\nProduct: {product_text}'
r = client.messages.create(model='claude-opus-4-5', max_tokens=400, messages=[{'role': 'user', 'content': prompt}])
print(json.loads(r.content[0].text))

선택적 필드 처리

prompt_optional = '''
Extract the data. For fields not present in the source text,
use null — do NOT omit the field.
Every field in the schema must appear in the output.

Schema:
{
  "company": "string",
  "ceo": "string or null",
  "founded": "YYYY or null",
  "revenue": "string or null",
  "employees": "number or null"
}

Text: Vertex AI Solutions is a B2B SaaS company.
'''

# Expected output: ceo, founded, revenue, employees all set to null
# NOT omitted — null fields are still present in the JSON
print(prompt_optional)

동일한 스키마를 사용한 여러 문서 추출

def extract_many(documents, schema):
    results = []
    for i, doc in enumerate(documents):
        try:
            r = client.messages.create(
                model='claude-opus-4-5', max_tokens=400,
                messages=[{'role': 'user', 'content': f'Extract data. Return JSON matching schema:\n{schema}\n\nDocument:\n{doc}'}]
            )
            parsed = json.loads(r.content[0].text)
            parsed['_source_doc'] = i
            parsed['_extraction_ok'] = True
            results.append(parsed)
        except (json.JSONDecodeError, Exception) as e:
            results.append({'_source_doc': i, '_extraction_ok': False, '_error': str(e)})
    return results

invoices = ['Invoice from Acme, March 2025, $500', 'Invoice from Beta Corp, April 2025, $1200']
results = extract_many(invoices, INVOICE_SCHEMA)
print(f'Processed: {len([r for r in results if r["_extraction_ok"]])} success, {len([r for r in results if not r["_extraction_ok"]])} failed')

추출 후 스키마 검증

def validate_extracted(data, required_fields, type_checks):
    errors = []

    # Check required fields
    for field in required_fields:
        if field not in data or data[field] is None:
            errors.append(f'Required field missing or null: {field}')

    # Check types
    for field, expected_type in type_checks.items():
        if field in data and data[field] is not None:
            if not isinstance(data[field], expected_type):
                errors.append(f'{field}: expected {expected_type.__name__}, got {type(data[field]).__name__}')

    return errors

extracted = {'invoice_number': 'INV-001', 'total_amount': 3255.0, 'vendor_name': 'Acme', 'invoice_date': '2025-03-15'}
required = ['invoice_number', 'total_amount', 'vendor_name']
types = {'total_amount': float, 'invoice_number': str, 'line_items': list}
errors = validate_extracted(extracted, required, types)
print('Validation errors:', errors)

반복적인 스키마 개선

스키마에 필드 설명 추가

ANNOTATED_SCHEMA = '''
{
  "invoice_number": "string // The unique identifier for this invoice, e.g., INV-2025-001",
  "invoice_date": "YYYY-MM-DD // Date the invoice was issued",
  "due_date": "YYYY-MM-DD or null // Payment due date; null if not specified",
  "subtotal": "number // Amount before tax, as a decimal number",
  "tax": "number or null // Tax amount as a decimal; null if tax is not listed",
  "total_amount": "number // Final amount to pay, including tax",
  "payment_terms": "string or null // e.g., Net 30, Due on receipt; null if not mentioned"
}
'''

print('Annotated schema adds context per field.')
print(f'Schema length: {len(ANNOTATED_SCHEMA)} chars')

추출된 필드의 신뢰도 점수

SCHEMA_WITH_CONFIDENCE = '''
{
  "fields": {
    "invoice_number": {"value": "string", "confidence": "high|medium|low"},
    "total_amount": {"value": "number", "confidence": "high|medium|low"},
    "due_date": {"value": "YYYY-MM-DD or null", "confidence": "high|medium|low"}
  },
  "overall_confidence": "high|medium|low",
  "extraction_notes": "string or null // Any ambiguities encountered"
}
'''

prompt = f'Extract invoice data with confidence scores.\nReturn JSON:\n{SCHEMA_WITH_CONFIDENCE}\n\nInvoice: Payment due within 30 days. Total is approximately $500.'
r = client.messages.create(model='claude-opus-4-5', max_tokens=300, messages=[{'role': 'user', 'content': prompt}])
result = json.loads(r.content[0].text)
print('Overall confidence:', result.get('overall_confidence'))
print('Notes:', result.get('extraction_notes'))

빠른 확인

스키마 기반 추출 — 핵심 요약