안녕하세요, 유저 여러분! 앞선 5탄의 그라운딩에 이어, 오늘은 대규모 문서 전처리 파이프라인의 완전히 최상단, 즉 비용과 속도를 모두 잡는 '문서 사전 분류(Triage)' 레이어 구축 이야기를 해보려고 합니다.
데이터 전처리 과정에서, 수많은 PDF 페이지를 아무 분류 없이 무조건 LLM이나 클라우드 OCR로 밀어 넣으면 리소스 낭비와 인프라 비용의 누수를 경험하게 되죠. 그렇기 때문에 "이 페이지는 빈 페이지인가? 스캔본인가? 표가 빽빽한가? OCR이나 LLM 호출이 필요한가?"를 먼저 영리하게 판단해 걸러주는 파이프라인이 필수적입니다.
이때 맨 앞단에서 가성비를 극대화해 주는 솔루션이 바로 PyMuPDF입니다. 픽셀 디코딩이나 외부 API 호출 없이, PDF 구조에 직접 접근하여 페이지당 단 몇 밀리초(ms) 만에 글자 수, 이미지 비율, 테이블 존재 여부 등의 시그널을 추출해 줍니다. 즉, API 토큰을 단 하나도 쓰기 전에 가장 효율적인 경로로 라우팅할 수 있는 '초가성비' 근거를 마련해 주는 것이죠.
자, 그럼 여러분의 서비스에 바로 적용할 수 있는 triage.py 전체 소스코드를 공개합니다!
"""
PyMuPDF 페이지 분류(Triage) — OCR 및 LLM 비용을 쓰기 전에 저렴하게 시그널을 추출하는 스크립트.
전략
----
각 페이지에서 비용이 거의 들지 않는 가벼운 시그널을 수집한 후, 다음 4가지 버킷 중 하나로 분류합니다:
SKIP — 빈 페이지 또는 의미 없는 페이지 (처리 제외)
TEXT_ONLY — 디지털 텍스트 추출 가능, 추가 OCR 불필요
OCR_NEEDED — 이미지가 대부분이거나 스캔된 페이지, OCR 엔진으로 전송
LLM_NEEDED — 의미론적 분석이 필요한 페이지 (양식, 혼합 레이아웃, 테이블 등)
비용 비교 (대략적인 상대적 수치)
PyMuPDF 시그널 추출: ~0.001x
OCR (예: Tesseract 또는 클라우드 OCR): ~1x
LLM (예: PyMuPDF Pro, GPT-4o, Claude 등): ~2–50x
"""
from dataclasses import dataclass, field
from enum import Enum, auto
from pathlib import Path
from typing import Optional
import pymupdf
# ── Triage buckets (분류 버킷 정의)
class Bucket(Enum):
SKIP = auto() # 빈 페이지 / 콘텐츠 거의 없음
TEXT_ONLY = auto() # 디지털 텍스트 기반, 추가 처리 불필요
OCR_NEEDED = auto() # 스캔본 또는 이미지 중심 페이지, OCR 필요
LLM_NEEDED = auto() # 복잡하고 의미론적 해석이 필요한 레이아웃
# ── Per-page signals (페이지별 시그널 데이터 구조)
@dataclass
class PageSignals:
page_number: int
width: float
height: float
# Text (텍스트 관련)
char_count: int = 0
word_count: int = 0
text_coverage: float = 0.0 # 텍스트 블록이 페이지 전체에서 차지하는 비율
has_native_text: bool = False
# Images (이미지 관련)
image_count: int = 0
image_coverage: float = 0.0 # 이미지가 페이지 전체에서 차지하는 비율
# Structure hints (구조적 힌트)
has_tables: bool = False
has_forms: bool = False # 위젯 주석(Annotation)을 통해 감지
block_count: int = 0
vector_drawing: bool = False # 이미지가 아닌 벡터 드로잉 커맨드 존재 여부
# Derived (분류 결과 파생 데이터)
bucket: Optional[Bucket] = field(default=None, init=False)
reason: str = field(default="", init=False)
# ── Signal extraction (시그널 추출 로직)
def extract_signals(page: pymupdf.Page) -> PageSignals:
"""
단일 PyMuPDF 페이지 객체로부터 저비용 시그널을 추출합니다.
모든 작업은 로컬 C/Python 레벨에서 처리되며 외부 서비스로 연동되지 않습니다.
"""
rect = page.rect
page_area = rect.width * rect.height or 1.0 # 0 나누기 방지 방어 코드
sig = PageSignals(
page_number=page.number, # 0부터 시작하는 페이지 번호
width=rect.width,
height=rect.height,
)
# Text (텍스트 추출)
# 글자/단어 수를 셀 때는 get_text("blocks")보다 get_text("words")가 더 빠릅니다.
words = page.get_text("words") # 결과 구조: (x0, y0, x1, y1, word, …)
sig.word_count = len(words)
sig.char_count = sum(len(w[4]) for w in words)
sig.has_native_text = sig.char_count > 20 # 단순 워터마크나 푸터는 무시
# Text spatial coverage via blocks (블록 데이터를 통한 텍스트의 공간적 커버리지 계산)
blocks = page.get_text("blocks") # (x0, y0, x1, y1, text, block_no, block_type)
sig.block_count = len(blocks)
text_area = sum(
(b[2] - b[0]) * (b[3] - b[1])
for b in blocks if b[6] == 0 # block_type 0 = text
)
sig.text_coverage = min(text_area / page_area, 1.0)
# Images (이미지 추출)
# get_image_info()는 픽셀 데이터를 직접 풀지 않고 bbox 데이터만 가져오므로 매우 가볍습니다.
images = page.get_image_info(hashes=False, xrefs=False)
sig.image_count = len(images)
img_area = sum(
(img["bbox"][2] - img["bbox"][0]) * (img["bbox"][3] - img["bbox"][1])
for img in images if img.get("bbox")
)
sig.image_coverage = min(img_area / page_area, 1.0)
# Tables (테이블 추출)
# PyMuPDF 내장 함수인 find_tables() 활용
tabs = page.find_tables()
sig.has_tables = len(tabs.tables) > 0
# Forms: widget annotations (checkboxes, text fields, dropdowns, etc.) (입력 양식 위젯 감지)
for annot in page.annots():
if annot.type[0] == pymupdf.PDF_ANNOT_WIDGET:
sig.has_forms = True
break
# Vector drawings: any path/curve drawing that is not an image. (벡터 드로잉 감지)
# get_drawings()는 스트로크/채우기 정보를 가볍게 반환합니다.
drawings = page.get_drawings()
sig.vector_drawing = len(drawings) > 0
return sig
# ── Triage rules ──────────────────────────────────────────────────────────────
def triage(sig: PageSignals,
*,
blank_char_threshold: int = 10,
blank_image_threshold: float = 0.02,
ocr_image_threshold: float = 0.25,
ocr_text_threshold: int = 30,
llm_complexity_score: int = 2) -> PageSignals:
"""
분류 규칙을 적용하고 시그널 객체에 최종 버킷과 사유(reason)를 기입합니다.
임계값(Threshold)들은 키워드 전용 인자로 제공되어 문서 성격에 맞게 튜닝이 가능합니다.
"""
chars = sig.char_count
imgcov = sig.image_coverage
# ── 규칙 1: SKIP — 빈 페이지 또는 콘텐츠가 거의 없는 경우
if chars < blank_char_threshold and imgcov < blank_image_threshold:
sig.bucket = Bucket.SKIP
sig.reason = f"빈 페이지 (글자 수={chars}, 이미지 비율={imgcov:.2f})"
return sig
# ── 규칙 2: OCR_NEEDED — 이미지가 지배적이며 디지털 텍스트가 거의 없는 경우
if imgcov >= ocr_image_threshold and chars < ocr_text_threshold:
sig.bucket = Bucket.OCR_NEEDED
sig.reason = (
f"이미지 중심 페이지 (이미지 비율={imgcov:.2f}, 글자 수={chars}) — "
"스캔본 또는 이미지 전용 페이지일 확률 높음"
)
return sig
# ── 규칙 3: LLM_NEEDED — 레이아웃이 복잡하거나 구조화된 콘텐츠인 경우
complexity = sum([
sig.has_tables,
sig.has_forms,
sig.image_count > 0 and sig.has_native_text, # 이미지와 텍스트 혼합
sig.block_count > 30, # 밀집된 레이아웃
sig.text_coverage < 0.10 and chars > 50, # 양식지와 같은 성긴 텍스트 배치
])
if complexity >= llm_complexity_score:
sig.bucket = Bucket.LLM_NEEDED
sig.reason = (
f"복잡한 레이아웃 (복잡도 점수={complexity}/5): "
+ ", ".join(filter(None, [
"테이블 존재" if sig.has_tables else "",
"양식(Form) 존재" if sig.has_forms else "",
"혼합 콘텐츠" if sig.image_count > 0 and sig.has_native_text else "",
f"밀집 블록({sig.block_count}개)" if sig.block_count > 30 else "",
"양식형 텍스트 구조" if sig.text_coverage < 0.10 and chars > 50 else "",
]))
)
return sig
# ── 규칙 4: TEXT_ONLY — 깨끗한 디지털 본문 텍스트만 있는 경우
sig.bucket = Bucket.TEXT_ONLY
sig.reason = (
f"일반 디지털 텍스트 (글자 수={chars}, 단어 수={sig.word_count}, "
f"텍스트 비율={sig.text_coverage:.2f})"
)
return sig
# ── Document-level triage (문서 레벨 종합 리포트 생성 구조)
@dataclass
class TriageReport:
path: str
page_count: int
results: list[PageSignals]
@property
def by_bucket(self) -> dict[Bucket, list[PageSignals]]:
out: dict[Bucket, list[PageSignals]] = {b: [] for b in Bucket}
for r in self.results:
out[r.bucket].append(r)
return out
def summary(self) -> str:
bb = self.by_bucket
lines = [
f"대상 문서 : {self.path}",
f"전체 페이지 : {self.page_count}",
"─" * 50,
]
for bucket in Bucket:
pages = bb[bucket]
if not pages:
continue
nums = ", ".join(str(p.page_number) for p in pages[:10])
if len(pages) > 10:
nums += f" … (+{len(pages)-10}장 더 있음)"
lines.append(f" {bucket.name:<12} {len(pages):>4} 페이지 [{nums}]")
lines.append("─" * 50)
total = self.page_count or 1
skip = len(bb[Bucket.SKIP])
lines.append(
f" 제외 가능한 페이지: {skip}/{total} ({skip/total*100:.0f}%) "
f"— 모든 페이지를 LLM에 보냈을 때 대비 절감된 예상 비용 수치입니다."
)
return "\n".join(lines)
def triage_document(
path: str | Path,
**triage_kwargs,
) -> TriageReport:
"""
PDF를 열어 모든 페이지를 분류하고 TriageReport 객체를 반환합니다.
읽기 전용으로 열리므로 원본 파일은 수정되지 않습니다.
"""
path = str(path)
doc = pymupdf.open(path)
results = []
for page in doc:
sig = extract_signals(page)
triage(sig, **triage_kwargs)
results.append(sig)
doc.close()
return TriageReport(path=path, page_count=len(results), results=results)
# ── Routing helpers (파이프라인 연동용 라우팅 헬퍼 함수들)
def route_pages(report: TriageReport) -> dict[str, list[int]]:
"""
0부터 시작하는 페이지 번호들을 각 처리 프로세스 경로별로 그룹화하여 반환합니다.
이 결과를 OCR / LLM 파이프라인에 그대로 넘겨주면 됩니다.
사용 예시
--------
routes = route_pages(report)
ocr_pages = routes["ocr"] # Tesseract / 클라우드 OCR로 전송
llm_pages = routes["llm"] # PyMuPDF Pro / GPT-4o / Claude 등으로 전송
text_pages = routes["text"] # page.get_text()로 무료 추출 가능
skip_pages = routes["skip"] # 완전히 제외하고 무시
"""
bb = report.by_bucket
return {
"skip": [p.page_number for p in bb[Bucket.SKIP]],
"text": [p.page_number for p in bb[Bucket.TEXT_ONLY]],
"ocr": [p.page_number for p in bb[Bucket.OCR_NEEDED]],
"llm": [p.page_number for p in bb[Bucket.LLM_NEEDED]],
}
def extract_text_pages(pdf_path: str | Path, page_indices: list[int]) -> dict[int, str]:
"""
TEXT_ONLY로 분류된 페이지에서 가볍고 빠르게 디지털 텍스트를 추출합니다.
결과 값: {페이지번호: 텍스트}
"""
doc = pymupdf.open(str(pdf_path))
out = {}
for i in page_indices:
out[i] = doc[i].get_text("text")
doc.close()
return out
def render_pages_for_ocr(
pdf_path: str | Path,
page_indices: list[int],
dpi: int = 200,
) -> dict[int, bytes]:
"""
OCR_NEEDED로 분류된 페이지를 지정된 DPI 속성의 PNG 바이트 데이터로 렌더링합니다.
결과 값: {페이지번호: png_bytes} -> 이 바이너리를 곧바로 OCR 엔진에 전달 가능합니다.
Tip: Tesseract OCR 엔진은 보통 150 DPI면 충분하며, 클라우드 API 기반 OCR은 300 DPI를 권장합니다.
"""
doc = pymupdf.open(str(pdf_path))
out = {}
mat = pymupdf.Matrix(dpi / 72, dpi / 72)
for i in page_indices:
pix = doc[i].get_pixmap(matrix=mat, colorspace=pymupdf.csGRAY)
out[i] = pix.tobytes("png")
doc.close()
return out
# ── CLI entry point (CLI 엔트리 포인트)
if __name__ == "__main__":
import sys
import json
if len(sys.argv) < 2:
print("Usage: python triage.py <file.pdf> [--details]")
sys.exit(1)
pdf_path = sys.argv[1]
show_details = "--details" in sys.argv
report = triage_document(pdf_path)
if show_details:
routes = route_pages(report)
output = json.dumps({
"path": report.path,
"page_count": report.page_count,
"routes": routes,
"details": [
{
"page": s.page_number,
"bucket": s.bucket.name,
"reason": s.reason,
"chars": s.char_count,
"words": s.word_count,
"images": s.image_count,
"image_cov": round(s.image_coverage, 3),
"text_cov": round(s.text_coverage, 3),
"has_tables": s.has_tables,
"has_forms": s.has_forms,
"has_vector": s.vector_drawing,
}
for s in report.results
]
}, indent=2, ensure_ascii=False)
print(output)
else:
print(report.summary())
print()
for s in report.results:
print(f" p{s.page_number:>4} [{s.bucket.name:<12}] {s.reason}")
핵심 요약 및 라우팅 시나리오
위 스크립트를 실제로 실행하면 페이지별 상태에 따라 유연하게 후처리를 태울 수 있는 로그 리포트가 출력됩니다.
p 0 [OCR_NEEDED ] 이미지 중심 페이지 (이미지 비율=1.00, 글자 수=14) — 스캔본 또는 이미지 전용 페이지일 확률 높음
p 1 [TEXT_ONLY ] 일반 디지털 텍스트 (글자 수=1362, 단어 수=93, 텍스트 비율=0.34)
p 2 [LLM_NEEDED ] 복잡한 레이아웃 (복잡도 점수=2/5): 테이블 존재, 혼합 콘텐츠
- p0 (이미지 중심 표지 등): 고비용 클라우드 OCR API 대신, 로컬 단에서 필수 키워드만 빠르게 스캔하고 패스하도록 가볍게 방어 코드를 태웁니다.
- p1 (디지털 본문): 무거운 인프라 리소스 없이 PyMuPDF의 내장
get_text()메서드로 텍스트 데이터를 비용 0원으로 즉시 긁어옵니다. - p2 (복잡한 양식/표): 구조가 성기거나 빽빽해 일반 파싱으로는 데이터가 깨지는 구역입니다. 이 페이지에 한해서만 AI 기반 레이아웃 파싱 솔루션인 **PyMuPDF Pro**을 조건부 투입합니다.
인사이트
대규모 데이터셋을 만지는 RAG 파이프라인 전처리 단에 이 사전 분류(Triage) 레이어를 한 번만 제대로 심어두면, 불필요한 고비용 인프라 호출을 완벽하게 통제할 수 있습니다. 시스템 최적화와 아키텍처 고도화를 고민 중이신 개발자분들께 작게나마 도움이 되었으면 좋겠습니다!
원문 참고: PyMuPDF로 구축하는 가성비 문서 분류(Triage) 파이프라인
지난 시리즈 보기: