비정형 텍스트에서 데이터 추출하기

정보 추출 문제

일반적인 추출 활용 사례

이메일 추출 파이프라인 구축

import openai
from pydantic import BaseModel
from typing import List, Optional
from enum import Enum

client = openai.OpenAI()

class Priority(str, Enum):
    urgent = 'urgent'
    high = 'high'
    normal = 'normal'
    low = 'low'

class EmailExtraction(BaseModel):
    subject_summary: str
    sender_intent: str
    product_mentioned: Optional[str]
    issue_category: str  # billing / technical / general / feedback
    priority: Priority
    action_required: bool
    action_description: Optional[str]
    customer_sentiment: str  # positive / negative / neutral / frustrated

def extract_from_email(email_body: str) -> EmailExtraction:
    result = client.beta.chat.completions.parse(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': 'You are an expert at analyzing customer emails and extracting structured information for a support team.'},
            {'role': 'user', 'content': f'Analyze this customer email:\n\n{email_body}'}
        ],
        response_format=EmailExtraction
    )
    return result.choices[0].message.parsed

LLM을 활용한 개체명 인식

import openai
from pydantic import BaseModel
from typing import List, Optional

client = openai.OpenAI()

class NamedEntity(BaseModel):
    text: str       # The exact text as it appears
    entity_type: str  # PERSON / ORG / LOCATION / DATE / MONEY / PRODUCT
    normalized: Optional[str]  # Standardized form where applicable

class NERResult(BaseModel):
    entities: List[NamedEntity]

text = '''
Apple Inc. CEO Tim Cook announced yesterday that the company will invest $1.2 billion
in a new manufacturing facility in Austin, Texas, expected to open in Q3 2026.
'''

result = client.beta.chat.completions.parse(
    model='gpt-4o-mini',
    messages=[
        {'role': 'system', 'content': 'Extract all named entities from the text. Classify each as PERSON, ORG, LOCATION, DATE, MONEY, or PRODUCT.'},
        {'role': 'user', 'content': text}
    ],
    response_format=NERResult
)
for entity in result.choices[0].message.parsed.entities:
    print(f'[{entity.entity_type}] {entity.text}')

대규모 문서 추출

import asyncio
import openai
from pydantic import BaseModel
from typing import List, Optional

async_client = openai.AsyncOpenAI()

class InvoiceExtraction(BaseModel):
    vendor: str
    total_amount: Optional[float]
    currency: str
    invoice_date: Optional[str]

async def extract_invoice(doc_text: str, semaphore: asyncio.Semaphore) -> InvoiceExtraction:
    async with semaphore:  # Limit concurrent requests
        result = await async_client.beta.chat.completions.parse(
            model='gpt-4o-mini',
            messages=[
                {'role': 'system', 'content': 'Extract invoice data.'},
                {'role': 'user', 'content': doc_text}
            ],
            response_format=InvoiceExtraction
        )
        return result.choices[0].message.parsed

async def process_invoices(documents: List[str]):
    sem = asyncio.Semaphore(5)  # Max 5 concurrent requests
    tasks = [extract_invoice(doc, sem) for doc in documents]
    return await asyncio.gather(*tasks, return_exceptions=True)

# results = asyncio.run(process_invoices(invoice_texts))
print('Async pipeline defined - handles rate limits via semaphore')

암시적·추론 정보 처리

추출 프롬프트 설계

복잡한 문서를 위한 다중 단계 추출

import openai
from pydantic import BaseModel
from typing import Optional

client = openai.OpenAI()

class DocumentType(BaseModel):
    doc_type: str  # invoice / contract / resume / report
    confidence: float

def classify_document(text: str) -> str:
    result = client.beta.chat.completions.parse(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': 'Classify the document type.'},
            {'role': 'user', 'content': text[:500]}  # Use only the beginning for classification
        ],
        response_format=DocumentType
    )
    return result.choices[0].message.parsed.doc_type

# Then route to the appropriate extraction schema
EXTRACTION_SCHEMAS = {
    'invoice': 'InvoiceSchema',  # Replace with actual Pydantic classes
    'contract': 'ContractSchema',
    'resume': 'ResumeSchema',
}

print('Multi-pass: classify first, then extract with the right schema')

추출 후 데이터 보강

추출 정확도 측정

지속적인 개선을 위한 추출 기록

빠른 확인

단원 요약