4장: 문서 이해와 OCR 통합

import anthropic
import base64
from pathlib import Path
 
client = anthropic.Anthropic()
 
pdf_data = base64.standard_b64encode(
    Path("report.pdf").read_bytes()
).decode("utf-8")
 
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "document",
                "source": {
                    "type": "base64",
                    "media_type": "application/pdf",
                    "data": pdf_data,
                },
            },
            {
                "type": "text",
                "text": """이 보고서를 분석하여 다음을 추출해주세요:
 
1. 보고서 제목, 저자, 날짜
2. 목차 구조
3. 핵심 요약 (500자 이내)
4. 주요 수치 데이터 (표 형식으로)
5. 결론 및 권고사항""",
            },
        ],
    }],
)

table_extraction_prompt = """이 문서 페이지에서 표(table)를 찾아 다음 형식으로 추출해주세요.
 
각 표에 대해:
1. 표 제목 또는 캡션
2. 열 헤더
3. 모든 행의 데이터
4. 병합된 셀이 있다면 명시
 
출력 형식 (JSON):
{
  "tables": [
    {
      "title": "표 제목",
      "headers": ["열1", "열2", "열3"],
      "rows": [
        ["값1", "값2", "값3"],
        ["값4", "값5", "값6"]
      ],
      "notes": "병합 셀, 특이사항 등"
    }
  ]
}"""

form_extraction_prompt = """이 양식(form) 이미지에서 모든 필드와 값을 추출해주세요.
 
각 필드에 대해:
- 필드 레이블 (label)
- 필드 값 (value) — 비어있으면 "empty"
- 필드 유형 (text, checkbox, radio, signature, date 등)
- 체크박스/라디오의 경우 선택 여부
 
JSON 형식으로 출력:
{
  "form_title": "양식 제목",
  "fields": [
    {"label": "이름", "value": "홍길동", "type": "text"},
    {"label": "동의", "value": true, "type": "checkbox"}
  ]
}"""

from dataclasses import dataclass
 
@dataclass
class DocumentPage:
    page_number: int
    text: str
    has_tables: bool
    has_images: bool
    confidence: float
 
class HybridDocumentProcessor:
    def __init__(self, ocr_engine, vlm_client):
        self.ocr = ocr_engine
        self.vlm = vlm_client
 
    def process_page(self, page_image: bytes) -> DocumentPage:
        # 1단계: OCR로 텍스트 추출 (빠르고 저렴)
        ocr_result = self.ocr.extract_text(page_image)
 
        # 2단계: 페이지 복잡도 판단
        has_tables = self._detect_tables(ocr_result)
        has_images = self._detect_visual_elements(page_image)
 
        # 3단계: 복잡한 페이지만 VLM으로 처리
        if has_tables or has_images or ocr_result.confidence < 0.85:
            vlm_result = self._analyze_with_vlm(page_image, ocr_result.text)
            return DocumentPage(
                page_number=ocr_result.page,
                text=vlm_result,
                has_tables=has_tables,
                has_images=has_images,
                confidence=0.95,
            )
 
        return DocumentPage(
            page_number=ocr_result.page,
            text=ocr_result.text,
            has_tables=False,
            has_images=False,
            confidence=ocr_result.confidence,
        )
 
    def _analyze_with_vlm(self, image: bytes, ocr_text: str) -> str:
        """VLM으로 복잡한 페이지 분석"""
        prompt = f"""이 문서 페이지를 분석해주세요.
OCR로 추출된 텍스트를 참고하되, 이미지에서 직접 읽은 정보를 우선합니다.
 
OCR 텍스트 (참고용):
{ocr_text}
 
작업:
1. 텍스트 내용을 정확하게 전사
2. 표가 있다면 마크다운 표로 변환
3. 차트/그래프가 있다면 데이터를 텍스트로 설명
4. 레이아웃 구조를 반영한 마크다운 출력"""
        # VLM API 호출...
        return "..."

contract_prompt = """이 계약서를 분석하여 다음 핵심 조항을 추출해주세요:
 
1. 계약 당사자 (갑/을)
2. 계약 기간
3. 계약 금액과 지급 조건
4. 해지 조건
5. 위약금 조항
6. 분쟁 해결 방법
7. 특이사항 또는 리스크 요소
 
각 항목에 해당하는 원문 조항 번호를 함께 인용해주세요."""

receipt_prompt = """이 영수증/인보이스에서 다음 정보를 추출하여 JSON으로 반환해주세요:
 
{
  "vendor": "판매자명",
  "date": "YYYY-MM-DD",
  "items": [
    {"name": "상품명", "quantity": 1, "unit_price": 10000, "total": 10000}
  ],
  "subtotal": 0,
  "tax": 0,
  "total": 0,
  "payment_method": "카드/현금",
  "currency": "KRW"
}
 
금액은 숫자로만, 통화 기호 제외. 읽기 어려운 부분은 "unclear"로 표시."""

paper_prompt = """이 학술 논문을 분석하여 구조화된 요약을 작성해주세요:
 
1. **메타데이터**: 제목, 저자, 소속, 발행일, DOI
2. **초록 요약**: 3줄 이내
3. **연구 질문/목적**: 핵심 연구 질문
4. **방법론**: 사용된 방법의 핵심
5. **주요 결과**: 정량적 결과 포함
6. **결론**: 핵심 기여
7. **한계**: 저자가 인정한 한계
8. **인용할 핵심 수치**: 표나 그래프의 주요 데이터"""

import asyncio
from typing import AsyncIterator
 
class BatchDocumentProcessor:
    def __init__(self, client, max_concurrent: int = 5):
        self.client = client
        self.semaphore = asyncio.Semaphore(max_concurrent)
 
    async def process_documents(
        self, documents: list[dict]
    ) -> AsyncIterator[dict]:
        """여러 문서를 병렬로 처리"""
        tasks = [
            self._process_single(doc) for doc in documents
        ]
 
        for completed in asyncio.as_completed(tasks):
            result = await completed
            yield result
 
    async def _process_single(self, document: dict) -> dict:
        async with self.semaphore:
            # 페이지별 처리
            pages = self._split_pages(document["data"])
            results = []
 
            for i, page in enumerate(pages):
                result = await self._analyze_page(page, i + 1)
                results.append(result)
 
            return {
                "filename": document["filename"],
                "pages": results,
                "total_pages": len(pages),
            }