저는 지난 3개월간 이커머스 플랫폼의 고객 서비스 AI 시스템을 구축하면서 가장 큰挑战은 바로 비정형 데이터의 구조화였습니다. 주문 확인 이메일, 배송 추적 이미지, 영수증 PDF가 매일 수천 건씩 유입되는 상황에서 수동 처리是不可能했습니다.
이 튜토리얼에서는 HolySheep AI를 활용하여 PDF, 이미지, 이메일에서 구조화된 데이터를 추출하는 방법을 실제 서비스에 적용한 경험과 함께 공유하겠습니다.
왜 AI 기반 데이터 추출인가?
전통적인 OCR(광학 문자 인식) 방식의 한계를 생각해봅시다. 표 형식이 섞인 PDF에서는 레이아웃 인식이 깨지고, 이미지의 폰트가 특수하면 인식률이 급격히 떨어집니다. 반면 AI 모델은 문맥을 이해하고 구조를 추론할 수 있습니다.
- 정확도: 레이아웃 복잡한 문서도 95% 이상 인식
- 속도: 문서당 2-3초 처리 ( humaines 대비 50배 이상)
- 유연성: 새 문서 형식에 빠르게 적응
1. 이미지에서 텍스트 추출
가장 기본적인 활용 사례부터 시작하겠습니다. 상품 이미지에서 브랜드, 모델명, 가격 정보를 추출하는 시스템을 구축했습니다.
import base64
import requests
import json
def encode_image_to_base64(image_path: str) -> str:
"""이미지 파일을 Base64로 인코딩"""
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
def extract_from_image(image_path: str, api_key: str) -> dict:
"""HolySheep AI Vision API로 이미지에서 구조화 데이터 추출"""
base64_image = encode_image_to_base64(image_path)
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": """이 상품 이미지에서 다음 정보를 JSON으로 추출해주세요:
- product_name: 상품명
- brand: 브랜드명
- price: 가격 (숫자만, 원 단위)
- currency: 통화 단위
- category: 상품 카테고리
응답은 반드시 유효한 JSON 객체로만 작성해주세요."""
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
"temperature": 0.1,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
# JSON 문자열을 파싱하여 반환
content = result['choices'][0]['message']['content']
return json.loads(content)
사용 예시
api_key = "YOUR_HOLYSHEEP_API_KEY"
try:
result = extract_from_image("product_image.jpg", api_key)
print(f"상품명: {result['product_name']}")
print(f"브랜드: {result['brand']}")
print(f"가격: {result['price']} {result['currency']}")
except Exception as e:
print(f"추출 실패: {e}")이 코드를 실행하면 보통 1.2-1.8초 내에 결과를 받을 수 있으며, 비용은 약 $0.0004 (약 0.5원) 수준입니다. 이미지당 이 가격이면 대량 처리에도 충분히 경제적입니다.
2. 이메일에서 주문 정보 추출
저의 실제 서비스에서는 매일 3,000건 이상의 주문 확인 이메일이 발송됩니다. 이제 이를 자동 분류하고 핵심 정보를 추출하는 시스템을 구축했습니다.
import requests
import re
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class OrderInfo:
order_id: str
customer_name: str
total_amount: float
currency: str
items: List[dict]
shipping_address: str
order_date: str
payment_method: str
def extract_order_from_email(email_body: str, api_key: str) -> OrderInfo:
"""이메일 본문에서 주문 정보를 구조화하여 추출"""
payload = {
"model": "gpt-4o-mini",
"messages": [
{
"role": "system",
"content": """당신은 이커머스 주문 정보 추출 전문가입니다.
이메일 본문에서 주문 정보를 정확히 추출하여 JSON으로 반환해주세요.
모든 필드는 반드시 채워야 하며, 정보가 없으면 null을 입력해주세요."""
},
{
"role": "user",
"content": f"""다음 이메일 본문에서 주문 정보를 추출해주세요:
{email_body}
반환 형식:
{{
"order_id": "주문번호",
"customer_name": "고객명",
"total_amount": 총금액숫자,
"currency": "KRW",
"items": [{{"name": "상품명", "quantity": 수량, "price": 가격}}],
"shipping_address": "배송지",
"order_date": "YYYY-MM-DD",
"payment_method": "결제수단"
}}"""
}
],
"temperature": 0.1,
"max_tokens": 1000,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload
)
data = response.json()
extracted = data['choices'][0]['message']['content']
# OrderInfo dataclass로 변환
order_dict = eval(extracted) # 또는 json.loads(extracted)
return OrderInfo(**order_dict)
배치 처리 예시
def process_email_batch(emails: List[str], api_key: str) -> List[OrderInfo]:
"""여러 이메일을 배치로 처리"""
results = []
for email in emails:
try:
order = extract_order_from_email(email, api_key)
results.append(order)
except Exception as e:
print(f"이메일 처리 실패: {e}")
continue
return results
사용 예시
api_key = "YOUR_HOLYSHEEP_API_KEY"
sample_email = """
[주문 확인] 2024-01-15 주문접수 안내
안녕하세요 홍길동님,
주문이 정상적으로 접수되었습니다.
주문번호: ORD-2024-0012345
주문일시: 2024년 1월 15일 14:30
결제수단: 신용카드(비씨카드)
주문상품:
- LG 스타일러 스팀 의류관리기 (1개) - 890,000원
- 올바른 구두クリ너 (2개) - 36,000원
배송지: 서울특별시 강남구 테헤란로 123building 5F
결제금액: 926,000원
감사합니다.
"""
order = extract_order_from_email(sample_email, api_key)
print(f"추출된 주문번호: {order.order_id}")
print(f"고객명: {order.customer_name}")
print(f"총액: {order.total_amount:,} {order.currency}")이메일 배치 처리 시 HolySheep AI의 gpt-4o-mini 모델을 사용하면 품질은 유지하면서 비용을 60% 절감할 수 있습니다. 1,000건 처리 시 약 $0.15 (약 200원) 수준입니다.
3. PDF 문서에서 데이터 추출
PDF는 복잡한 레이아웃과 표, 다단 편집 등으로 가장 어려운 형식입니다. HolySheep AI Vision API를 활용하면 멀티모달 처리로这些问题을 해결할 수 있습니다.
import requests
import json
from pathlib import Path
def extract_from_pdf(pdf_path: str, api_key: str) -> dict:
"""PDF 파일에서 구조화된 데이터 추출"""
# PDF를 이미지로 변환 후 처리
# PyMuPDF 또는 pdf2image 라이브러리로 페이지별 이미지 추출
try:
import fitz # PyMuPDF
doc = fitz.open(pdf_path)
all_extracted_data = []
for page_num in range(len(doc)):
page = doc[page_num]
pix = page.get_pixmap(dpi=200)
image_bytes = pix.tobytes("png")
base64_image = base64.b64encode(image_bytes).decode('utf-8')
# HolySheep AI로 이미지 분석
result = analyze_pdf_page(base64_image, page_num + 1, api_key)
all_extracted_data.append(result)
doc.close()
return {"pages": all_extracted_data}
except ImportError:
print("PyMuPDF 설치 필요: pip install pymupdf")
raise
def analyze_pdf_page(base64_image: str, page_num: int, api_key: str) -> dict:
"""단일 PDF 페이지를 분석하여 데이터 추출"""
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": f"""이 PDF 페이지 (페이지 {page_num})에서 다음 정보를 추출해주세요:
1. 제목과 주요 헤딩
2. 핵심 데이터 테이블 (있다면)
3. 주요 수치나 통계
4. 날짜 및 참조 번호
결과는 구조화된 JSON으로 반환해주세요."""
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{base64_image}"
}
}
]
}
],
"temperature": 0.1,
"max_tokens": 2000,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=60
)
return response.json()['choices'][0]['message']['content']
인보이스 PDF 처리 예시
def process_invoice(invoice_path: str, api_key: str) -> dict:
"""인보이스 PDF에서 청구 정보 추출"""
import fitz
doc = fitz.open(invoice_path)
page = doc[0]
pix = page.get_pixmap(dpi=300)
base64_image = base64.b64encode(pix.tobytes("png")).decode('utf-8')
doc.close()
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": """이 인보이스 문서에서 다음 정보를 정확히 추출해주세요:
- invoice_number: 청구서 번호
- invoice_date: 청구일 (YYYY-MM-DD)
- due_date: 납기일 (YYYY-MM-DD)
- vendor_name: 공급자명
- vendor_address: 공급자 주소
- customer_name: 구매자명
- customer_address: 구매자 주소
- items: [{"description": "품목", "quantity": 수량, "unit_price": 단가, "total": 합계}]
- subtotal: 소계
- tax: 세액
- total_amount: 총액
- payment_terms: 결제 조건
금액은 숫자로, 날짜는 YYYY-MM-DD 형식으로 반환해주세요."""
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{base64_image}"
}
}
]
}
],
"temperature": 0,
"max_tokens": 1500,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload
)
return json.loads(response.json()['choices'][0]['message']['content'])
사용 예시
api_key = "YOUR_HOLYSHEEP_API_KEY"
try:
invoice_data = process_invoice("invoice.pdf", api_key)
print(f"청구서 번호: {invoice_data['invoice_number']}")
print(f"총액: {invoice_data['total_amount']:,.0f}원")
print(f"품목 수: {len(invoice_data['items'])}개")
except Exception as e:
print(f"인보이스 처리 오류: {e}")4. 실제 RAG 시스템과의 통합
제가 구축한 기업 내부 문서 RAG 시스템에서는 위에서 추출한 데이터를 벡터 데이터베이스에 저장하여 검색 가능하게 만들었습니다.
import requests
import json
from datetime import datetime
def create_structured_document(raw_data: dict, doc_type: str) -> dict:
"""추출된 원시 데이터를 RAG용 구조화 문서로 변환"""
# 문서 타입별 프롬프트 선택
prompts = {
"invoice": "인보이스 정보를 자연스러운 검색 가능한 텍스트로 변환",
"email": "이메일 정보를 검색 가능한 요약으로 변환",
"image": "이미지 내용을 텍스트 설명으로 변환"
}
payload = {
"model": "gpt-4o-mini",
"messages": [
{
"role": "system",
"content": "당신은 문서 구조화 전문가입니다. 데이터를 검색 가능한 자연어로 변환해주세요."
},
{
"role": "user",
"content": f"""{prompts.get(doc_type, '데이터')}를 수행해주세요.
입력 데이터:
{json.dumps(raw_data, ensure_ascii=False, indent=2)}
요구사항:
1. 핵심 정보를 요약 (100자 이내)
2. 검색에 유리한 키워드 추출
3. 전체 내용을 자연어로 기술
4. JSON 형식으로 반환
"""
}
],
"temperature": 0.3,
"max_tokens": 800,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload
)
result = json.loads(response.json()['choices'][0]['message']['content'])
return {
"doc_type": doc_type,
"raw_data": raw_data,
"summary": result.get("summary", ""),
"keywords": result.get("keywords", []),
"content": result.get("content", ""),
"created_at": datetime.now().isoformat(),
"embedding_version": "v1"
}
def index_to_vector_db(document: dict, pinecone_client):
"""구조화 문서를 벡터 DB에 인덱싱"""
# OpenAI Embedding으로 벡터 변환
embedding_response = requests.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "text-embedding-3-small",
"input": document["content"]
}
)
vector = embedding_response.json()['data'][0]['embedding']
# Pinecone에 업로드
pinecone_client.upsert(
vectors=[{
"id": f"doc_{document['created_at']}",
"values": vector,
"metadata": {
"doc_type": document["doc_type"],
"summary": document["summary"],
"keywords": document["keywords"],
"created_at": document["created_at"]
}
}]
)
return True
전체 파이프라인 실행
def full_extraction_pipeline(source: str, source_type: str, api_key: str):
"""완전한 데이터 추출 파이프라인"""
if source_type == "pdf":
data = extract_from_pdf(source, api_key)
elif source_type == "email":
data = extract_order_from_email(source, api_key)
elif source_type == "image":
data = extract_from_image(source, api_key)
else:
raise ValueError(f"지원하지 않는 소스 타입: {source_type}")
# RAG용 문서 변환
structured_doc = create_structured_document(data, source_type)
return structured_doc
사용 예시
api_key = "YOUR_HOLYSHEEP_API_KEY"
result = full_extraction_pipeline("invoice.pdf", "pdf", api_key)
print(f"문서 요약: {result['summary']}")
print(f"추출된 키워드: {result['keywords']}")비용 최적화 전략
실제 운영에서는 비용 최적화가 매우 중요합니다. HolySheep AI의 다양한 모델을 상황에 맞게 선택해야 합니다.
| 작업 유형 | 권장 모델 | 비용 ($/1K 토큰) | 처리 시간 |
|---|---|---|---|
| 고품질 이미지 분석 | gpt-4o | 입력: $2.50 / 출력: $10.00 | 1.5-2.5초 |
| 대량 배치 처리 | gpt-4o-mini | 입력: $0.15 / 출력: $0.60 | 0.8-1.5초 |
| 복잡한 PDF 표 | claude-sonnet-4-20250514 | 입력: $3.00 / 출력: $15.00 | 2.0-3.5초 |
| 간단한 텍스트 추출 | deepseek-chat | 입력: $0.14 / 출력: $0.28 | 0.5-1.0초 |
제 경험상 gpt-4o-mini로 80%의 작업을 처리하고, 정밀도가 필요한 20%에만 gpt-4o를 사용하면 비용을 70% 절감하면서도 품질을 유지할 수 있었습니다.
자주 발생하는 오류와 해결책
1. Base64 인코딩 오류
# ❌ 잘못된 방법: 파일 경로로 직접 전달
"image_url": {"url": "file:///path/to/image.jpg"}
✅ 올바른 방법: Base64 데이터 URL 형식
import base64
with open("image.jpg", "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
"image_url": {"url": f"data:image/jpeg;base64,{image_data}"}
이미지 타입별 MIME 타입 참고:
PNG: "image/png"
JPEG: "image/jpeg"
WebP: "image/webp"
GIF: "image/gif"
2. 토큰 제한 초과 오류
# ❌ 큰 PDF를 한 번에 처리할 때 발생
PDFsms 10MB 이상 또는 50페이지 이상
✅ 해결 방법 1: 페이지별 분할 처리
def process_large_pdf(pdf_path: str, api_key: str):
import fitz
doc = fitz.open(pdf_path)
all_results = []
for page_num in range(0, len(doc), 5): # 5페이지씩 처리
pages = []
for i in range(page_num, min(page_num + 5, len(doc))):
page = doc[i]
pix = page.get_pixmap(dpi=150) # DPI 낮추기
pages.append(pix.tobytes("png"))
# 배치로 처리
result = process_page_batch(pages, api_key)
all_results.extend(result)
return all_results
✅ 해결 방법 2: DPI 최적화