실제 개발 환경에서의 첫 번째 장애
지난주 팀에서 문서 자동 분류 시스템을 구축하던 중 치명적인 오류를 만났습니다. 수백 개의 PDF 문서를 처리하는 배치 잡이 갑자기
ConnectionResetError: [Errno 104] Connection reset by peer로 중단된 것입니다. 로그를 분석해보니 타임아웃 설정 없이 대용량 PDF 파일을 순차적으로 처리하다가 서버 연결이 끊긴 것이었죠.
이 튜토리얼에서는 **HolySheep AI**를 활용하여 다양한 문서 형식(PDF, Word, PPT)을 안정적으로 텍스트로 변환하고, 변환된 텍스트를 LLM으로 분석하는 실전 파이프라인을 구축하는 방법을 설명드리겠습니다.
1. 필요한 라이브러리 설치
# 문서 파싱 및 AI API 연동을 위한 필수 패키지
pip install python-docx PyPDF2 python-pptx requests tenacity pillow
각 라이브러리 역할
python-docx : .docx 파일 읽기
PyPDF2 : .pdf 파일 읽기 (pdfminer 권장)
python-pptx : .pptx 파일 읽기
requests : HolySheep AI API 호출
tenacity : 재시도 로직 (타이아웃 처리)
pillow : 이미지 추출 (스캔 PDF용)
# 완전한 requirements.txt 예시
python-docx==1.1.0
PyPDF2==3.0.1
python-pptx==1.0.1
requests==2.31.0
tenacity==8.2.3
pillow==10.2.0
pdfminer.six==20231228
2. HolySheep AI API 기본 설정
import os
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
재시도 데코레이터를 통한 안정적 API 호출
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_holysheep_api(prompt, model="gpt-4.1", max_tokens=2000):
"""
HolySheep AI API를 통해 문서 분석 요청
모델별 가격:
- gpt-4.1: $8/MTok (8센트/1K 토큰)
- claude-sonnet-4-20250514: $15/MTok (15센트/1K 토큰)
- gemini-2.5-flash: $2.50/MTok (2.5센트/1K 토큰)
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": "당신은 문서를 분석하는 전문가입니다."},
{"role": "user", "content": prompt}
],
"max_tokens": max_tokens,
"temperature": 0.3
}
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
if response.status_code == 401:
raise ValueError("API 키가 유효하지 않습니다. HolySheep AI 대시보드에서 확인하세요.")
elif response.status_code == 429:
raise ValueError("요청 한도를 초과했습니다. 잠시 후 재시도하세요.")
elif response.status_code != 200:
raise ValueError(f"API 오류: {response.status_code} - {response.text}")
return response.json()["choices"][0]["message"]["content"]
3. PDF 파싱 모듈 구현
제가 실무에서 가장 많이 마주치는 문제는 **스캔 PDF와 텍스트 PDF의 처리 방식 차이**입니다. 스캔된 문서는 OCR이 필요하고, 텍스트 기반 문서는 직접 텍스트 추출이 가능합니다.
import io
from PyPDF2 import PdfReader
from pdfminer.high_level import extract_text as pdfminer_extract_text
from pdfminer.layout import LAParams
def extract_text_from_pdf(file_path, use_advanced=True):
"""
PDF 파일에서 텍스트 추출
Args:
file_path: PDF 파일 경로
use_advanced: True면 pdfminer 사용 (복잡한 레이아웃 대응)
Returns:
str: 추출된 텍스트
"""
extracted_text = ""
try:
if use_advanced:
# pdfminer.six: 복잡한 레이아웃, 표, 다단 레이아웃에更强
extracted_text = pdfminer_extract_text(
file_path,
laparams=LAParams(
line_margin=0.5,
word_margin=0.1,
char_margin=2,
detect_vertical=True
)
)
else:
# PyPDF2: 간단한 텍스트 PDF에 적합
with open(file_path, 'rb') as f:
reader = PdfReader(f)
for page in reader.pages:
text = page.extract_text()
if text:
extracted_text += text + "\n\n"
# 텍스트 정제
extracted_text = clean_extracted_text(extracted_text)
if not extracted_text.strip():
print(f"⚠️ {file_path}: 텍스트가 비어있습니다. 스캔 문서일 수 있습니다.")
return None
print(f"✅ {file_path}: {len(extracted_text)}자 추출 완료")
return extracted_text
except Exception as e:
print(f"❌ PDF 파싱 오류 ({file_path}): {type(e).__name__}: {str(e)}")
return None
def clean_extracted_text(text):
"""추출된 텍스트 정제"""
import re
# 연속된 공백 제거
text = re.sub(r'\s+', ' ', text)
# 불필요한 줄바꿈 정리
text = re.sub(r'[\n\r]+', '\n', text)
# 특수문자 정리
text = re.sub(r'[^\w\s가-힣.,!?;:()\[\]{}""\'\'\n-]', '', text)
return text.strip()
4. Word(.docx) 문서 파싱
from docx import Document
def extract_text_from_docx(file_path):
"""
Word(.docx) 파일에서 텍스트 추출
표, 머리글/바닥글, 주석도 포함하여 추출
"""
try:
doc = Document(file_path)
full_text = []
# 본문 단락 추출
for para in doc.paragraphs:
if para.text.strip():
full_text.append(para.text)
# 표에서 텍스트 추출 (표가 많은 문서에서 중요)
for table in doc.tables:
for row in table.rows:
row_text = []
for cell in row.cells:
if cell.text.strip():
row_text.append(cell.text.strip())
if row_text:
full_text.append(" | ".join(row_text))
# 헤더/푸터 추출
for section in doc.sections:
for header in section.header.paragraphs:
if header.text.strip():
full_text.append(f"[머리글] {header.text}")
for footer in section.footer.paragraphs:
if footer.text.strip():
full_text.append(f"[바닥글] {footer.text}")
extracted_text = "\n".join(full_text)
if not extracted_text.strip():
print(f"⚠️ {file_path}: 텍스트가 비어있습니다.")
return None
print(f"✅ {file_path}: {len(extracted_text)}자, {len(doc.paragraphs)}단락 추출 완료")
return extracted_text
except Exception as e:
print(f"❌ Word 파싱 오류 ({file_path}): {type(e).__name__}: {str(e)}")
return None
5. PowerPoint(.pptx) 파싱
from pptx import Presentation
from pptx.util import Inches, Pt
def extract_text_from_pptx(file_path, include_slide_numbers=True):
"""
PowerPoint(.pptx) 파일에서 텍스트 추출
슬라이드별 구조를 유지하여 분석 용이성 향상
"""
try:
prs = Presentation(file_path)
slides_text = []
for slide_num, slide in enumerate(prs.slides, 1):
slide_content = []
# 슬라이드 제목
if slide.shapes.title:
title = slide.shapes.title.text.strip()
if title:
slide_content.append(f"## 슬라이드 {slide_num}: {title}")
# 모든 텍스트 Shape 탐색
for shape in slide.shapes:
if hasattr(shape, "text_frame"):
for paragraph in shape.text_frame.paragraphs:
text = paragraph.text.strip()
if text and text not in [s for s in slide_content]:
slide_content.append(text)
# 표 내용 추출
if shape.has_table:
table = shape.table
for row in table.rows:
row_cells = [cell.text.strip() for cell in row.cells]
if any(row_cells):
slide_content.append(" | ".join(row_cells))
if slide_content:
slides_text.append("\n".join(slide_content))
final_text = "\n\n---\n\n".join(slides_text)
if not final_text.strip():
print(f"⚠️ {file_path}: 텍스트가 비어있습니다.")
return None
print(f"✅ {file_path}: {len(final_text)}자, {len(prs.slides)}슬라이드 추출 완료")
return final_text
except Exception as e:
print(f"❌ PowerPoint 파싱 오류 ({file_path}): {type(e).__name__}: {str(e)}")
return None
6. 문서 분류 및 요약 파이프라인
import os
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class DocumentResult:
file_path: str
file_type: str
extracted_text: Optional[str]
category: Optional[str]
summary: Optional[str]
word_count: int
error: Optional[str] = None
def classify_and_summarize_document(text, file_name):
"""
HolySheep AI를 활용하여 문서 분류 및 요약
"""
if not text or len(text) < 100:
return {"category": "unknown", "summary": "텍스트 부족"}
# 토큰 비용 최적화: 긴 문서는 앞부분만 분석
MAX_CHARS = 8000
analysis_text = text[:MAX_CHARS] if len(text) > MAX_CHARS else text
prompt = f"""다음 문서를 분석하여 분류(category)와 요약(summary)을 제공하세요.
문서명: {file_name}
---
{analysis_text}
---
응답 형식 (JSON):
{{"category": "분류명", "summary": "3-5문장 요약"}}
분류 옵션: 계약서, 보고서, 기술문서, 마케팅자료, 재무제표, 인사문서, 일반문서"""
try:
response = call_holysheep_api(
prompt=prompt,
model="gpt-4.1", # 비용 효율적인 모델 선택
max_tokens=500
)
# JSON 파싱
import json
import re
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
result = json.loads(json_match.group())
return result
else:
return {"category": "unknown", "summary": response[:200]}
except Exception as e:
print(f"분류/요약 중 오류: {e}")
return {"category": "error", "summary": str(e)}
def process_single_document(file_path):
"""단일 문서 처리"""
path = Path(file_path)
ext = path.suffix.lower()
result = DocumentResult(
file_path=str(file_path),
file_type=ext,
extracted_text=None,
category=None,
summary=None,
word_count=0
)
# 파일 형식별 분기
if ext == '.pdf':
result.extracted_text = extract_text_from_pdf(file_path)
elif ext in ['.docx', '.doc']:
result.extracted_text = extract_text_from_docx(file_path)
elif ext in ['.pptx', '.ppt']:
result.extracted_text = extract_text_from_pptx(file_path)
else:
result.error = f"지원하지 않는 파일 형식: {ext}"
return result
if result.extracted_text:
result.word_count = len(result.extracted_text)
# AI 분석 수행
analysis = classify_and_summarize_document(
result.extracted_text,
path.name
)
result.category = analysis.get("category")
result.summary = analysis.get("summary")
return result
def batch_process_documents(directory_path, max_workers=3):
"""
디렉토리의 모든 문서를 배치 처리
Args:
directory_path: 문서가 있는 디렉토리 경로
max_workers: 동시 처리 스레드 수 (너무 많으면 API 제한 발생)
"""
doc_dir = Path(directory_path)
supported_extensions = {'.pdf', '.docx', '.doc', '.pptx', '.ppt'}
# 지원 파일 목록 수집
files_to_process = [
f for f in doc_dir.rglob('*')
if f.is_file() and f.suffix.lower() in supported_extensions
]
print(f"📁 {len(files_to_process)}개 파일 발견")
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_file = {
executor.submit(process_single_document, str(f)): f
for f in files_to_process
}
for future in as_completed(future_to_file):
file_path = future_to_file[future]
try:
result = future.result()
results.append(result)
except Exception as e:
print(f"❌ 처리 실패 ({file_path}): {e}")
results.append(DocumentResult(
file_path=str(file_path),
file_type=file_path.suffix,
extracted_text=None,
category=None,
summary=None,
word_count=0,
error=str(e)
))
# 결과 요약
successful = [r for r in results if not r.error]
failed = [r for r in results if r.error]
print(f"\n📊 처리 결과: {len(successful)}성공 / {len(failed)}실패")
return results
7. 사용 예제 및 실행
# 최종 사용 예제
if __name__ == "__main__":
# HolySheep AI API 키 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# 단일 파일 테스트
test_file = "./sample_report.pdf"
if os.path.exists(test_file):
print("=== 단일 파일 테스트 ===")
result = process_single_document(test_file)
print(f"파일: {result.file_path}")
print(f"분류: {result.category}")
print(f"요약: {result.summary}")
print(f"글자수: {result.word_count:,}자")
# 배치 처리 테스트
batch_directory = "./documents"
if os.path.exists(batch_directory):
print("\n=== 배치 처리 ===")
results = batch_process_documents(batch_directory, max_workers=2)
# 분류별 통계
from collections import Counter
categories = [r.category for r in results if r.category]
category_counts = Counter(categories)
print("\n📈 분류 통계:")
for cat, count in category_counts.most_common():
print(f" - {cat}: {count}건")
# 비용 추정 (gpt-4.1: $8/MTok)
total_chars = sum(r.word_count for r in results)
estimated_tokens = total_chars // 4 # 대략적인 토큰 추정
estimated_cost = (estimated_tokens / 1_000_000) * 8
print(f"\n💰 예상 비용 (gpt-4.1): ${estimated_cost:.4f}")
자주 발생하는 오류와 해결책
오류 1: ConnectionResetError: [Errno 104] Connection reset by peer
원인: 대용량 PDF 순차 처리 시 서버 연결 타임아웃
해결:
# 방법 1: requests 세션 사용 및 타임아웃 설정
session = requests.Session()
session.mount('https://', requests.adapters.HTTPAdapter(
max_retries=3,
pool_connections=10,
pool_maxsize=20
))
방법 2: tenacity를 통한 재시도 로직 적용
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
retry=retry_if_exception_type((requests.exceptions.ConnectionError,
requests.exceptions.Timeout)),
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=4, max=30)
)
def robust_api_call(payload, timeout=120):
response = session.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=timeout
)
return response
오류 2: 401 Unauthorized - Invalid API Key
원인: HolySheep AI API 키 누락, 만료, 또는 형식 오류
해결:
# API 키 검증 함수 추가
def validate_api_key():
"""HolySheep AI API 키 유효성 검사"""
if not HOLYSHEEP_API_KEY:
raise ValueError("HOLYSHEEP_API_KEY 환경변수가 설정되지 않았습니다.")
if HOLYSHEEP_API_KEY == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("API 키를 실제 HolySheep AI 키로 교체하세요.")
# 키 형식 검증 (HolySheep AI는 sk-로 시작)
if not HOLYSHEEP_API_KEY.startswith("sk-"):
raise ValueError(f"API 키 형식이 올바르지 않습니다: {HOLYSHEEP_API_KEY[:10]}...")
# 실제 연결 테스트
headers = {"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"}
test_response = requests.get(
f"{HOLYSHEEP_BASE_URL}/models",
headers=headers,
timeout=10
)
if test_response.status_code == 401:
raise ValueError("API 키가 만료되었거나 유효하지 않습니다. 대시보드에서 확인하세요.")
return True
사용 전 검증
validate_api_key()
오류 3: MemoryError 또는 OSError: [Errno 28] No space left on device
원인: 대용량 문서 배치 처리 중 메모리/디스크 부족
해결:
import gc
import psutil
def process_with_memory_management(file_path, chunk_size=5):
"""
메모리 관리와 함께 문서 처리
- 청크 단위 처리
- 주기적 가비지 컬렉션
"""
process = psutil.Process()
# 메모리 사용량 체크
memory_mb = process.memory_info().rss / 1024 / 1024
if memory_mb > 500: # 500MB 이상 사용 시
print(f"⚠️ 메모리 사용량 높음: {memory_mb:.1f}MB, 가비지 컬렉션 실행")
gc.collect()
# 디스크 공간 체크
disk_usage = psutil.disk_usage('/')
if disk_usage.percent > 90:
raise OSError("디스크 공간이 부족합니다. ({:.1f}% 사용)".format(disk_usage.percent))
try:
result = process_single_document(file_path)
return result
finally:
# 처리 완료 후 메모리 정리
del result
gc.collect()
배치 처리 시 메모리 관리 적용
for batch_start in range(0, len(files), chunk_size):
batch = files[batch_start:batch_start + chunk_size]
for file_path in batch:
try:
result = process_with_memory_management(file_path)
results.append(result)
except MemoryError:
print(f"❌ 메모리 부족으로 건너뜀: {file_path}")
continue
# 배치 완료 후 정리
gc.collect()
print(f"📦 배치 {batch_start//chunk_size + 1} 완료, 메모리 정리됨")
오류 4: UnicodeDecodeError: 'utf-8' codec can't decode byte
원인: PDF 내 이미지나 특수 인코딩된 텍스트
해결:
def safe_extract_with_fallback(file_path):
"""
여러 인코딩 시도 후 실패 시 기본값 반환
"""
# 1차: pdfminer로 시도
try:
text = pdfminer_extract_text(file_path)
if text and len(text.strip()) > 50:
return text
except Exception as e:
print(f"pdfminer 실패: {e}")
# 2차: PyPDF2로 시도
try:
with open(file_path, 'rb') as f:
reader = PdfReader(f)
texts = []
for page in reader.pages:
try:
texts.append(page.extract_text())
except Exception:
texts.append("")
combined = "\n".join(texts)
if combined.strip():
return combined
except Exception as e:
print(f"PyPDF2 실패: {e}")
# 3차: OCR 필요 안내
print(f"⚠️ 텍스트 추출 실패. 스캔 문서일 가능성이 있습니다: {file_path}")
return "[스캔 문서 - OCR 필요]"
인코딩 안전한 텍스트 처리
extracted = safe_extract_with_fallback(file_path)
if extracted:
# 인코딩 오류 방지를 위한 정규화
import unicodedata
normalized = unicodedata.normalize('NFKC', extracted)
clean_text = normalized.encode('utf-8', errors='replace').decode('utf-8')
비용 최적화 팁
HolySheep AI를 활용할 때 비용을 절감하려면 다음 전략을 고려하세요:
- 적절한 모델 선택: 단순 분류 작업에는
gemini-2.5-flash($2.50/MTok)가 GPT-4.1($8/MTok) 대비 3배 저렴합니다
- 토큰 절약: 긴 문서는 앞부분 8,000자만 분석에 사용 (LLM은 문맥 초반을 더 중요하게 처리)
- 배치 처리: 여러 파일을 동시 처리하여 API 호출 오버헤드 감소
- 캐싱: 동일 문서 재처리 시 이전 결과를 활용
- 프리미엄 모델: Claude Sonnet 4.5($15/MTok)는 더 정확한 분석이 필요한 계약서/법률문서에 권장
결론
문서 파싱 파이프라인 구축 시 가장 중요한 세 가지는 **안정적인 연결 처리**, **메모리 관리**, **적절한 오류 복구 로직**입니다. HolySheep AI의 글로벌 API 게이트웨이를 활용하면 다양한 모델을 단일 API 키로 통합 관리하면서 비용 최적화까지 달성할 수 있습니다.
특히 HolySheep AI는 해외 신용카드 없이 로컬 결제가 가능하여, 개발자 친화적인 결제 옵션을 제공합니다. 현재 [무료 크레딧 제공 중](https://www.holysheep.ai/register)이니 바로 시작해보세요!
👉
HolySheep AI 가입하고 무료 크레딧 받기