5장: 비정형 데이터에서 구조화된 정보 추출

import fitz  # PyMuPDF
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
 
 
class InvoiceData(BaseModel):
    """송장 데이터"""
    invoice_number: str = Field(description="송장 번호")
    date: str = Field(description="발행일 (YYYY-MM-DD)")
    vendor_name: str = Field(description="공급자 상호")
    vendor_registration_number: str = Field(description="사업자등록번호")
    total_amount: float = Field(description="합계 금액")
    tax_amount: float = Field(description="세액")
    items: list["InvoiceItem"] = Field(description="품목 목록")
 
 
class InvoiceItem(BaseModel):
    """송장 품목"""
    description: str = Field(description="품목명")
    quantity: int = Field(ge=1, description="수량")
    unit_price: float = Field(description="단가")
    amount: float = Field(description="금액")
 
 
def extract_text_from_pdf(pdf_path: str) -> str:
    """PDF에서 텍스트를 추출합니다."""
    doc = fitz.open(pdf_path)
    text_parts = []
    for page in doc:
        text_parts.append(page.get_text())
    return "\n".join(text_parts)
 
 
def extract_invoice_data(pdf_path: str) -> InvoiceData:
    """PDF 송장에서 구조화된 데이터를 추출합니다."""
    text = extract_text_from_pdf(pdf_path)
 
    client = instructor.from_openai(OpenAI())
    result = client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=InvoiceData,
        max_retries=2,
        messages=[
            {
                "role": "system",
                "content": (
                    "다음 텍스트는 송장(Invoice) PDF에서 추출한 내용입니다. "
                    "송장의 모든 정보를 정확히 추출하세요. "
                    "금액은 숫자만 포함하고 통화 기호는 제외하세요."
                )
            },
            {"role": "user", "content": text}
        ]
    )
    return result

import fitz
import base64
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
 
 
def pdf_pages_to_images(pdf_path: str) -> list[str]:
    """PDF 각 페이지를 base64 이미지로 변환합니다."""
    doc = fitz.open(pdf_path)
    images = []
    for page in doc:
        pix = page.get_pixmap(dpi=200)
        img_bytes = pix.tobytes("png")
        b64 = base64.b64encode(img_bytes).decode("utf-8")
        images.append(b64)
    return images
 
 
def extract_from_scanned_pdf(pdf_path: str) -> InvoiceData:
    """스캔 PDF에서 멀티모달 LLM으로 데이터를 추출합니다."""
    images = pdf_pages_to_images(pdf_path)
 
    client = instructor.from_openai(OpenAI())
 
    # 이미지를 Vision API에 전달
    content = [
        {
            "type": "text",
            "text": "이 송장 이미지에서 모든 정보를 추출하세요."
        }
    ]
    for img_b64 in images:
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/png;base64,{img_b64}",
                "detail": "high"
            }
        })
 
    result = client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=InvoiceData,
        max_retries=2,
        messages=[
            {
                "role": "system",
                "content": "송장 이미지에서 정확한 데이터를 추출하세요."
            },
            {"role": "user", "content": content}
        ]
    )
    return result

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
import base64
 
 
class ReceiptItem(BaseModel):
    """영수증 품목"""
    name: str = Field(description="상품명")
    quantity: int = Field(ge=1, description="수량")
    price: float = Field(description="가격")
 
 
class ReceiptData(BaseModel):
    """영수증 데이터"""
    store_name: str = Field(description="매장명")
    date: str = Field(description="구매일 (YYYY-MM-DD)")
    items: list[ReceiptItem] = Field(description="구매 품목 목록")
    subtotal: float = Field(description="소계")
    tax: float = Field(description="부가세")
    total: float = Field(description="합계")
    payment_method: str = Field(description="결제 수단 (카드/현금/기타)")
 
 
def extract_receipt(image_path: str) -> ReceiptData:
    """영수증 이미지에서 데이터를 추출합니다."""
    with open(image_path, "rb") as f:
        img_b64 = base64.b64encode(f.read()).decode("utf-8")
 
    client = instructor.from_openai(OpenAI())
 
    return client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=ReceiptData,
        max_retries=2,
        messages=[
            {
                "role": "system",
                "content": (
                    "영수증 이미지에서 모든 정보를 정확히 추출하세요. "
                    "금액은 숫자만 포함합니다. "
                    "날짜 형식이 다른 경우 YYYY-MM-DD로 변환하세요."
                )
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/png;base64,{img_b64}",
                            "detail": "high"
                        }
                    }
                ]
            }
        ]
    )

import httpx
from bs4 import BeautifulSoup
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
 
 
class ProductInfo(BaseModel):
    """상품 정보"""
    name: str = Field(description="상품명")
    price: float = Field(description="가격 (원)")
    original_price: float | None = Field(
        default=None,
        description="할인 전 원래 가격 (할인이 없으면 null)"
    )
    rating: float | None = Field(
        default=None,
        description="평점 (5점 만점)"
    )
    review_count: int = Field(default=0, description="리뷰 수")
    description: str = Field(description="상품 설명 요약")
    specifications: dict[str, str] = Field(
        default_factory=dict,
        description="주요 사양 (키-값 쌍)"
    )
 
 
def extract_product_from_url(url: str) -> ProductInfo:
    """웹페이지에서 상품 정보를 추출합니다."""
    # HTML 가져오기
    response = httpx.get(url, follow_redirects=True)
    soup = BeautifulSoup(response.text, "html.parser")
 
    # 불필요한 요소 제거
    for tag in soup.find_all(["script", "style", "nav", "footer"]):
        tag.decompose()
 
    # 메인 콘텐츠 텍스트 추출
    text = soup.get_text(separator="\n", strip=True)
    # 텍스트 길이 제한 (토큰 절약)
    text = text[:5000]
 
    client = instructor.from_openai(OpenAI())
 
    return client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=ProductInfo,
        messages=[
            {
                "role": "system",
                "content": (
                    "다음은 상품 페이지에서 추출한 텍스트입니다. "
                    "상품 정보를 정확히 추출하세요. "
                    "정보를 찾을 수 없는 필드는 null로 표시하세요."
                )
            },
            {"role": "user", "content": text}
        ]
    )

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
 
 
class TableCell(BaseModel):
    """테이블 셀"""
    value: str = Field(description="셀 값")
    is_header: bool = Field(default=False, description="헤더 셀 여부")
 
 
class ExtractedTable(BaseModel):
    """추출된 테이블"""
    title: str = Field(description="테이블 제목 (유추 가능한 경우)")
    headers: list[str] = Field(description="열 헤더 목록")
    rows: list[list[str]] = Field(description="데이터 행 (각 행은 문자열 목록)")
    row_count: int = Field(description="데이터 행 수 (헤더 제외)")
 
 
class DocumentTables(BaseModel):
    """문서에서 추출한 모든 테이블"""
    tables: list[ExtractedTable] = Field(description="추출된 테이블 목록")
    total_tables: int = Field(description="총 테이블 수")
 
 
def extract_tables(document_text: str) -> DocumentTables:
    """문서 텍스트에서 테이블을 추출합니다."""
    client = instructor.from_openai(OpenAI())
 
    return client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=DocumentTables,
        messages=[
            {
                "role": "system",
                "content": (
                    "다음 문서에서 모든 테이블을 추출하세요. "
                    "테이블이 텍스트 형태로 표현되어 있더라도 구조를 파악하여 추출합니다. "
                    "각 테이블의 헤더와 데이터 행을 분리하세요."
                )
            },
            {"role": "user", "content": document_text}
        ]
    )

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Literal
 
 
class Entity(BaseModel):
    """추출된 엔티티"""
    name: str = Field(description="엔티티 이름")
    entity_type: Literal[
        "person", "organization", "location",
        "product", "event", "date", "amount"
    ] = Field(description="엔티티 유형")
    description: str = Field(description="엔티티에 대한 간단한 설명")
 
 
class Relation(BaseModel):
    """엔티티 간 관계"""
    source: str = Field(description="주체 엔티티 이름")
    relation_type: str = Field(description="관계 유형 (예: CEO_OF, LOCATED_IN)")
    target: str = Field(description="대상 엔티티 이름")
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="관계 추출 신뢰도"
    )
 
 
class KnowledgeGraph(BaseModel):
    """추출된 지식 그래프"""
    entities: list[Entity] = Field(description="추출된 엔티티 목록")
    relations: list[Relation] = Field(description="엔티티 간 관계 목록")
 
 
def extract_knowledge_graph(text: str) -> KnowledgeGraph:
    """텍스트에서 엔티티와 관계를 추출하여 지식 그래프를 구성합니다."""
    client = instructor.from_openai(OpenAI())
 
    return client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=KnowledgeGraph,
        messages=[
            {
                "role": "system",
                "content": (
                    "텍스트에서 모든 주요 엔티티를 식별하고, "
                    "엔티티 간의 관계를 추출하세요. "
                    "관계 유형은 UPPER_SNAKE_CASE로 작성하세요. "
                    "확실하지 않은 관계는 confidence를 낮게 설정하세요."
                )
            },
            {"role": "user", "content": text}
        ]
    )
 
 
# 사용 예시
text = """
삼성전자는 2024년 서울 강남구에 새로운 AI 연구소를 설립했습니다.
이재용 회장이 직접 개소식에 참석했으며, 초대 소장으로 김박사가 임명되었습니다.
이 연구소는 Google DeepMind와 공동 연구 협약을 체결할 예정입니다.
"""
 
kg = extract_knowledge_graph(text)
for entity in kg.entities:
    print(f"[{entity.entity_type}] {entity.name}: {entity.description}")
for rel in kg.relations:
    print(f"  {rel.source} --{rel.relation_type}--> {rel.target} ({rel.confidence:.0%})")

import asyncio
import instructor
from openai import AsyncOpenAI
from pydantic import BaseModel, Field
 
 
class ContractKeyTerms(BaseModel):
    """계약서 핵심 조건"""
    contract_type: str = Field(description="계약 유형")
    parties: list[str] = Field(description="계약 당사자")
    effective_date: str = Field(description="계약 시작일")
    expiration_date: str | None = Field(description="계약 종료일")
    total_value: float | None = Field(description="계약 총액")
    key_obligations: list[str] = Field(description="주요 의무사항")
    termination_conditions: list[str] = Field(description="해지 조건")
 
 
async def extract_single_contract(
    client: instructor.AsyncInstructor,
    document_text: str,
    doc_id: str
) -> tuple[str, ContractKeyTerms]:
    """단일 계약서에서 핵심 조건을 추출합니다."""
    result = await client.chat.completions.create(
        model="gpt-4o-2026-02",
        response_model=ContractKeyTerms,
        max_retries=2,
        messages=[
            {
                "role": "system",
                "content": "계약서에서 핵심 조건을 추출하세요."
            },
            {"role": "user", "content": document_text}
        ]
    )
    return doc_id, result
 
 
async def extract_batch_contracts(
    documents: dict[str, str]
) -> dict[str, ContractKeyTerms]:
    """여러 계약서를 비동기로 동시 추출합니다."""
    client = instructor.from_openai(AsyncOpenAI())
 
    tasks = [
        extract_single_contract(client, text, doc_id)
        for doc_id, text in documents.items()
    ]
 
    results = await asyncio.gather(*tasks, return_exceptions=True)
 
    extracted = {}
    for result in results:
        if isinstance(result, Exception):
            print(f"추출 실패: {result}")
        else:
            doc_id, data = result
            extracted[doc_id] = data
 
    return extracted