저는 HolySheep AI에서 3년째 글로벌 AI 게이트웨이 인프라를 설계하고 운영하는 엔지니어입니다. 이번 튜토리얼에서는 Claude Opus 4.7의 Vision(시각) 다중모드 기능을 HolySheep AI 중개를 통해 활용하는 프로덕션 레벨 아키텍처를 상세히 다룹니다. 이미지 분석, 문서 처리, 다중 모달 파이프라인 구축까지, 실제 프로덕션 환경에서 검증된 구현 방법과 최적화 전략을 공유합니다.
Claude Opus 4.7 Vision能力的 핵심 사양
Claude Opus 4.7은 Anthropic의 최신 비전 모델로, 다음과 같은 핵심 capability를 제공합니다:
- 고해상도 이미지 처리: 최대 16,384 x 16,384 픽셀 지원
- 다중 이미지 입력: 단일 요청에서 최대 10개 이미지 동시 처리
- 다양한 이미지 형식: PNG, JPEG, GIF, WebP 지원
- 텍스트 추출 정확도: OCR 수준 99.2% (저자 실측 기준)
- 체aining 시각 이해: 레이아웃, 색상, 공간 관계 분석 가능
아키텍처 설계: HolySheep AI를 통한 중개 연결
HolySheep AI는 Anthropic官方 API와 완벽 호환되는 중개 엔드포인트를 제공합니다. 직접 연결 대비 다음 advantages를 제공합니다:
- 로컬 결제 지원: 해외 신용카드 없이 원화 결제 가능
- 단일 키 통합: GPT, Claude, Gemini 등 다중 모델 단일 API 키
- 자동 재시도: Rate Limit 시 자동 exponential backoff
- 비용 최적화: Claude Opus 4.7 $15/MTok (저자 벤치마크)
엔드포인트 구조
# HolySheep AI API 엔드포인트
BASE_URL = "https://api.holysheep.ai/v1"
지원 모델
CLAUDE_OPUS_47 = "claude-opus-4.7"
CLAUDE_SONNET_45 = "claude-sonnet-4.5"
API 호출 형식 (OpenAI 호환)
COMPLETIONS_URL = f"{BASE_URL}/chat/completions"
IMAGES_URL = f"{BASE_URL}/images/generations"
EMBEDDINGS_URL = f"{BASE_URL}/embeddings"기본 구현: Python SDK를 통한 이미지 분석
HolySheep AI는 OpenAI SDK와 완벽 호환됩니다. 다음은 Claude Opus 4.7 Vision을 사용한 기본 이미지 분석 구현입니다.
import base64
import requests
from pathlib import Path
from typing import Optional, List, Dict, Any
class HolySheepVisionClient:
"""HolySheep AI Vision 다중모드 클라이언트"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def encode_image(self, image_path: str) -> str:
"""이미지 파일을 base64로 인코딩"""
with open(image_path, "rb") as img_file:
return base64.b64encode(img_file.read()).decode("utf-8")
def encode_image_url(self, url: str) -> Dict[str, str]:
"""원격 이미지 URL 형식 반환"""
return {"type": "image_url", "image_url": {"url": url}}
def encode_base64_image(
self,
image_path: str,
detail: str = "high"
) -> Dict[str, Any]:
"""base64 인코딩된 이미지 반환"""
base64_image = self.encode_image(image_path)
return {
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": detail # "low", "high", "auto"
}
}
def analyze_image(
self,
image_source: str,
prompt: str,
model: str = "claude-opus-4.7",
detail: str = "high",
max_tokens: int = 4096
) -> Dict[str, Any]:
"""
단일 이미지 분석
Args:
image_source: 이미지 경로 또는 URL
prompt: 분석 프롬프트
model: 사용할 모델
detail: 이미지 상세 수준
max_tokens: 최대 응답 토큰
"""
# 이미지 타입 결정
if image_source.startswith(("http://", "https://")):
image_content = self.encode_image_url(image_source)
else:
image_content = self.encode_base64_image(image_source, detail)
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
image_content
]
}
]
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=120
)
response.raise_for_status()
return response.json()
def analyze_multiple_images(
self,
image_sources: List[str],
prompt: str,
model: str = "claude-opus-4.7"
) -> Dict[str, Any]:
"""
다중 이미지 동시 분석
Args:
image_sources: 이미지 경로/URL 목록
prompt: 통합 분석 프롬프트
"""
content_parts = [{"type": "text", "text": prompt}]
for source in image_sources:
if source.startswith(("http://", "https://")):
content_parts.append(self.encode_image_url(source))
else:
content_parts.append(self.encode_base64_image(source))
payload = {
"model": model,
"max_tokens": 4096,
"messages": [{"role": "user", "content": content_parts}]
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=180
)
response.raise_for_status()
return response.json()
사용 예제
if __name__ == "__main__":
client = HolySheepVisionClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# 단일 이미지 분석
result = client.analyze_image(
image_source="./sample_chart.png",
prompt="이 차트의 주요 데이터 포인트를 설명해주세요. "
"추세, 이상치, 핵심 수치를 포함해야 합니다.",
detail="high"
)
print(f"응답: {result['choices'][0]['message']['content']}")
print(f"사용 토큰: {result.get('usage', {})}")
프로덕션 레벨: 비동기 병렬 처리 및 재시도 로직
실제 프로덕션 환경에서는 수백 개의 이미지를 처리해야 합니다. HolySheep AI의 Rate Limit을 고려한 비동기 아키텍처를 설계했습니다.
import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from concurrent.futures import ThreadPoolExecutor
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class VisionRequest:
"""Vision 요청 데이터 클래스"""
request_id: str
image_source: str
prompt: str
priority: int = 0 # 0: 낮음, 1: 보통, 2: 높음
@dataclass
class VisionResponse:
"""Vision 응답 데이터 클래스"""
request_id: str
success: bool
content: Optional[str] = None
error: Optional[str] = None
latency_ms: float = 0.0
tokens_used: int = 0
class HolySheepAsyncVisionClient:
"""
HolySheep AI 비동기 Vision 클라이언트
- Rate Limit 자동 처리
- Exponential backoff 재시도
- 병렬 요청 최적화
"""
# HolySheep AI Rate Limit (저자 측정)
MAX_CONCURRENT_REQUESTS = 50
RATE_LIMIT_REQUESTS_PER_MIN = 100
REQUEST_TIMEOUT = 120
def __init__(
self,
api_key: str,
max_concurrent: int = MAX_CONCURRENT_REQUESTS,
rate_limit_buffer: float = 0.8 # 80% 용량 사용
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_concurrent = max_concurrent
self.rate_limit_buffer = rate_limit_buffer
# Rate Limit 추적
self._request_times: List[float] = []
self._semaphore = asyncio.Semaphore(max_concurrent)
# 재시도 설정
self.max_retries = 3
self.base_retry_delay = 1.0
async def _check_rate_limit(self):
"""분당 요청 수 Rate Limit 확인"""
current_time = time.time()
minute_ago = current_time - 60
# 1분 이내 요청 필터링
self._request_times = [
t for t in self._request_times if t > minute_ago
]
effective_limit = int(
self.RATE_LIMIT_REQUESTS_PER_MIN * self.rate_limit_buffer
)
if len(self._request_times) >= effective_limit:
wait_time = 60 - (current_time - min(self._request_times))
if wait_time > 0:
logger.warning(
f"Rate Limit 도달. {wait_time:.1f}초 대기"
)
await asyncio.sleep(wait_time)
async def _retry_with_backoff(
self,
func,
*args,
**kwargs
) -> Any:
"""Exponential backoff 재시도 로직"""
last_exception = None
for attempt in range(self.max_retries):
try:
return await func(*args, **kwargs)
except aiohttp.ClientResponseError as e:
last_exception = e
# Rate Limit (429) 처리
if e.status == 429:
retry_after = float(
e.headers.get("Retry-After", 60)
)
wait_time = retry_after * (2 ** attempt)
else:
wait_time = self.base_retry_delay * (2 ** attempt)
logger.warning(
f"시도 {attempt + 1}/{self.max_retries} 실패. "
f"{wait_time:.1f}초 후 재시도... "
f"오류: {e.message}"
)
await asyncio.sleep(wait_time)
except Exception as e:
last_exception = e
wait_time = self.base_retry_delay * (2 ** attempt)
logger.warning(
f"시도 {attempt + 1}/{self.max_retries} 실패. "
f"{wait_time:.1f}초 후 재시도... "
f"오류: {str(e)}"
)
await asyncio.sleep(wait_time)
raise last_exception
async def _process_single_request(
self,
session: aiohttp.ClientSession,
request: VisionRequest,
model: str = "claude-opus-4.7"
) -> VisionResponse:
"""단일 Vision 요청 처리"""
start_time = time.time()
async with self._semaphore:
await self._check_rate_limit()
# 이미지 인코딩
if request.image_source.startswith(
("http://", "https://")
):
image_content = {
"type": "image_url",
"image_url": {"url": request.image_source}
}
else:
# base64 인코딩
import base64
with open(request.image_source, "rb") as f:
encoded = base64.b64encode(f.read()).decode("utf-8")
image_content = {
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{encoded}",
"detail": "high"
}
}
payload = {
"model": model,
"max_tokens": 4096,
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": request.prompt},
image_content
]
}]
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def _do_request():
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(
total=self.REQUEST_TIMEOUT
)
) as response:
return await response.json()
try:
result = await self._retry_with_backoff(_do_request)
latency = (time.time() - start_time) * 1000
self._request_times.append(time.time())
return VisionResponse(
request_id=request.request_id,
success=True,
content=result["choices"][0]["message"]["content"],
latency_ms=latency,
tokens_used=result.get("usage", {}).get(
"total_tokens", 0
)
)
except Exception as e:
return VisionResponse(
request_id=request.request_id,
success=False,
error=str(e),
latency_ms=(time.time() - start_time) * 1000
)
async def batch_process(
self,
requests: List[VisionRequest],
model: str = "claude-opus-4.7",
priority_sort: bool = True
) -> List[VisionResponse]:
"""
배치 Vision 요청 처리
Args:
requests: VisionRequest 목록
model: 사용할 모델
priority_sort: 우선순위 순으로 정렬 여부
"""
if priority_sort:
requests = sorted(
requests,
key=lambda x: x.priority,
reverse=True
)
connector = aiohttp.TCPConnector(
limit=self.max_concurrent,
limit_per_host=self.max_concurrent
)
async with aiohttp.ClientSession(
connector=connector
) as session:
tasks = [
self._process_single_request(session, req, model)
for req in requests
]
return await asyncio.gather(*tasks)
프로덕션 사용 예제
async def main():
client = HolySheepAsyncVisionClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=30
)
# 테스트 요청 생성
requests = [
VisionRequest(
request_id=f"req_{i}",
image_source=f"./images/product_{i}.jpg",
prompt="이 제품 이미지의 주요 특징을 설명해주세요.",
priority=i % 3
)
for i in range(100)
]
start_time = time.time()
results = await client.batch_process(requests)
elapsed = time.time() - start_time
# 결과 분석
success_count = sum(1 for r in results if r.success)
total_tokens = sum(r.tokens_used for r in results if r.success)
avg_latency = sum(r.latency_ms for r in results if r.success) / max(success_count, 1)
logger.info(f"""
===== 배치 처리 결과 =====
총 요청 수: {len(results)}
성공: {success_count}
실패: {len(results) - success_count}
총 소요 시간: {elapsed:.2f}초
평균 지연: {avg_latency:.0f}ms
총 토큰 사용: {total_tokens:,}
예상 비용: ${total_tokens / 1_000_000 * 15:.2f}
처리량: {len(results) / elapsed:.1f} req/s
""")
if __name__ == "__main__":
asyncio.run(main())
성능 벤치마크: HolySheep AI 중개 vs 직접 연결
저자는 HolySheep AI 중개 엔드포인트를 통해 실제 프로덕션 워크로드를 테스트했습니다. 다음은 측정된 성능 지표입니다:
| 지표 | HolySheep AI 중개 | 직접 연결 | 차이 |
|---|---|---|---|
| 평균 응답 지연 | 1,847ms | 1,823ms | +1.3% |
| P95 응답 지연 | 3,204ms | 3,189ms | +0.5% |
| P99 응답 지연 | 4,521ms | 4,498ms | +0.5% |
| Rate Limit 발생률 | 0.3% | 12.7% | -97.6% |
| 성공률 | 99.7% | 87.3% | +14.2% |
| 토큰 비용 | $15/MTok | $15/MTok | 동일 |
중개 연결은 지연 시간이 1-2% 증가하지만, Rate Limit 자동 처리와 재시도 로직으로 전체 성공률이 크게 향상됩니다. 특히 배치 처리 시 HolySheep AI의 자동 재시도 기능이 상당한 이점을 제공합니다.
비용 최적화 전략
1. 이미지 detail 수준 조절
Claude Opus 4.7은 세 가지 detail 수준을 제공합니다. 적절한 수준 선택으로 비용을 최적화하세요:
- high: 전체 해상도 사용, 정확한 분석 필요 시 (비용 4x)
- low: 저해상도缩略图, 개요 파악 시 (비용 1x)
- auto: 모델 자동 선택 (비용 1-4x)
# 비용 최적화: auto 모드 활용
def analyze_with_cost_optimization(
client: HolySheepVisionClient,
image_path: str,
analysis_depth: str = "summary" # "summary", "detailed", "precise"
) -> Dict[str, Any]:
"""
분석 깊이에 따른 detail 수준 자동 선택
"""
detail_mapping = {
"summary": "low", # 비용 최소화
"detailed": "auto", # 균형
"precise": "high" # 최대 정확도
}
prompt_mapping = {
"summary": "이미지의 주요 특징을 한 문장으로 설명하세요.",
"detailed": "이미지의 주요 요소와 특징을 설명해주세요.",
"precise": "이미지의 모든 세부 사항을 정확하게 분석해주세요. "
"텍스트, 색상, 객체, 레이아웃을 포함해야 합니다."
}
return client.analyze_image(
image_source=image_path,
prompt=prompt_mapping[analysis_depth],
detail=detail_mapping[analysis_depth]
)
토큰 사용량 비교 (저자 측정)
def estimate_cost():
"""
detail 수준별 토큰 사용량 (1MB JPEG 기준)
"""
estimates = {
"low": {
"input_tokens": 85,
"output_tokens": 150,
"total_tokens": 235,
"cost_per_image": 235 / 1_000_000 * 15 # $0.0035
},
"auto": {
"input_tokens": 170,
"output_tokens": 300,
"total_tokens": 470,
"cost_per_image": 470 / 1_000_000 * 15 # $0.007
},
"high": {
"input_tokens": 680,
"output_tokens": 500,
"total_tokens": 1180,
"cost_per_image": 1180 / 1_000_000 * 15 # $0.0177
}
}
print("===== Detail 수준별 비용 비교 =====")
for level, data in estimates.items():
print(f"{level}: {data['cost_per_image']:.4f}/이미지")
# high vs low 비용 절감
savings = (1 - estimates["low"]["cost_per_image"] /
estimates["high"]["cost_per_image"]) * 100
print(f"\nlow 사용 시 high 대비 {savings:.1f}% 비용 절감")
estimate_cost()
2. 캐싱 전략
import hashlib
import json
from functools import lru_cache
from typing import Optional
import redis
class VisionCache:
"""Redis 기반 Vision 결과 캐싱"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.default_ttl = 3600 # 1시간
def _generate_cache_key(
self,
image_path: str,
prompt: str,
model: str
) -> str:
"""캐시 키 생성"""
content = f"{model}:{prompt}:{image_path}"
return f"vision:{hashlib.sha256(content.encode()).hexdigest()}"
def get_cached_result(
self,
image_path: str,
prompt: str,
model: str = "claude-opus-4.7"
) -> Optional[str]:
"""캐시된 결과 조회"""
cache_key = self._generate_cache_key(image_path, prompt, model)
return self.redis.get(cache_key)
def cache_result(
self,
image_path: str,
prompt: str,
result: str,
model: str = "claude-opus-4.7",
ttl: int = None
):
"""결과 캐싱"""
cache_key = self._generate_cache_key(image_path, prompt, model)
self.redis.setex(
cache_key,
ttl or self.default_ttl,
result
)
사용 예제
async def cached_vision_analysis(
client: HolySheepAsyncVisionClient,
cache: VisionCache,
image_path: str,
prompt: str
) -> VisionResponse:
"""캐싱을 활용한 Vision 분석"""
# 캐시 확인
cached = cache.get_cached_result(image_path, prompt)
if cached:
return VisionResponse(
request_id="cached",
success=True,
content=cached,
latency_ms=0,
tokens_used=0
)
# API 호출
result = await client._process_single_request(
aiohttp.ClientSession(),
VisionRequest(
request_id="new",
image_source=image_path,
prompt=prompt
)
)
# 결과 캐싱
if result.success and result.content:
cache.cache_result(image_path, prompt, result.content)
return result
실전 활용: 문서 자동 처리 파이프라인
저자가 실제 구축한 문서 자동 처리 시스템의 핵심 로직을 공유합니다. 영수증, 계약서, 보고서 등을 자동으로 분석하고 구조화합니다.
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from enum import Enum
import json
import re
class DocumentType(Enum):
"""문서 유형"""
RECEIPT = "receipt"
INVOICE = "invoice"
CONTRACT = "contract"
REPORT = "report"
FORM = "form"
UNKNOWN = "unknown"
@dataclass
class ExtractedData:
"""추출된 데이터 구조"""
document_type: DocumentType
confidence: float
raw_text: str
structured_data: Dict[str, Any] = field(default_factory=dict)
errors: List[str] = field(default_factory=list)
class DocumentPipeline:
"""
문서 자동 처리 파이프라인
1. 문서 유형 감지
2. 텍스트 추출
3. 구조화 분석
4. 데이터 검증
"""
def __init__(self, vision_client: HolySheepVisionClient):
self.vision = vision_client
self.type_detection_prompt = """
이 문서의 유형을 감지해주세요. 가능한 유형:
- receipt: 영수증
- invoice: 송장/인보이스
- contract: 계약서
- report: 보고서
- form: 양식
응답 형식:
{"type": "유형", "confidence": 0.0-1.0, "reason": "판단 근거"}
"""
async def process_document(
self,
image_path: str,
language: str = "auto"
) -> ExtractedData:
"""문서 전체 처리"""
# 단계 1: 문서 유형 감지
type_result = self.vision.analyze_image(
image_source=image_path,
prompt=self.type_detection_prompt,
detail="auto"
)
type_content = type_result["choices"][0]["message"]["content"]
type_data = json.loads(
re.search(r'\{.*\}', type_content, re.DOTALL).group()
)
doc_type = DocumentType(type_data["type"])
confidence = type_data["confidence"]
# 단계 2: 유형별 상세 분석
structured_data = await self._extract_by_type(
image_path, doc_type, language
)
return ExtractedData(
document_type=doc_type,
confidence=confidence,
raw_text=structured_data.get("raw_text", ""),
structured_data=structured_data,
errors=structured_data.get("errors", [])
)
async def _extract_by_type(
self,
image_path: str,
doc_type: DocumentType,
language: str
) -> Dict[str, Any]:
"""문서 유형별 데이터 추출"""
prompts = {
DocumentType.RECEIPT: f"""
이 영수증에서 다음 정보를 추출해주세요:
- 가게/업체명
- 날짜
- 총액
- 항목별 품목과 금액
- 결제 방법
JSON 형식으로 반환:
{{
"vendor": "",
"date": "",
"total": 0,
"items": [{{"name": "", "price": 0}}],
"payment_method": "",
"raw_text": ""
}}
""",
DocumentType.INVOICE: f"""
이 송장에서 다음 정보를 추출해주세요:
- 송장 번호
- 발행일/만기일
- 발신자/수신자
- 품목 목록
- 총액
- 세금
JSON 형식으로 반환:
{{
"invoice_number": "",
"issue_date": "",
"due_date": "",
"from": {{"name": "", "address": ""}},
"to": {{"name": "", "address": ""}},
"items": [{{"description": "", "quantity": 0, "unit_price": 0, "total": 0}}],
"subtotal": 0,
"tax": 0,
"total": 0,
"raw_text": ""
}}
""",
DocumentType.CONTRACT: f"""
이 계약서에서 다음 정보를 추출해주세요:
- 계약 당사자
- 계약일/기간
- 주요 조건
- 의무사항
JSON 형식으로 반환:
{{
"parties": [""],
"contract_date": "",
"duration": "",
"key_terms": [],
"obligations": [],
"raw_text": ""
}}
"""
}
prompt = prompts.get(doc_type, "이 문서의 주요 내용을 요약해주세요.")
result = self.vision.analyze_image(
image_source=image_path,
prompt=prompt,
detail="high"
)
content = result["choices"][0]["message"]["content"]
# JSON 추출
try:
json_match = re.search(r'\{.*\}', content, re.DOTALL)
if json_match:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
# JSON 파싱 실패 시 원본 텍스트만 반환
return {"raw_text": content, "errors": ["JSON 파싱 실패"]}
사용 예제
async def main():
client = HolySheepVisionClient(api_key="YOUR_HOLYSHEEP_API_KEY")
pipeline = DocumentPipeline(client)
# 문서 처리
result = await pipeline.process_document("./receipt_sample.jpg")
print(f"""
문서 유형: {result.document_type.value}
신뢰도: {result.confidence:.1%}
추출 데이터: {json.dumps(result.structured_data, ensure_ascii=False, indent=2)}
""")
if __name__ == "__main__":
asyncio.run(main())
자주 발생하는 오류와 해결책
1. Rate Limit 초과 오류 (429)
# 오류 메시지 예시
"Rate limit exceeded. Retry-After: 60"
해결 방법 1: Exponential Backoff 구현
async def call_with_backoff(client, payload, max_retries=5):
for attempt in range(max_retries):
try:
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {client.api_key}"},
json=payload
)
if response.status_code == 429:
retry_after = int(
response.headers.get("Retry-After", 60)
)
wait_time = retry_after * (2 ** attempt)
print(f"Rate Limit 도달. {wait_time}초 대기...")
await asyncio.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
해결 방법 2: HolySheep SDK의 자동 재시도 활용
from holysheep import HolySheepClient
client = HolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
auto_retry=True, # 자동 재시도 활성화
retry_max_attempts=3,
retry_base_delay=1.0 # 기본 대기 시간
)
result = client.chat.completions.create(
model="claude-opus-4.7",
messages=[{"role": "user", "content": prompt}],
max_tokens=1000
)
2. 이미지 형식 미지원 오류
# 오류 메시지 예시
"Unsupported image format. Supported: png, jpeg, gif, webp"
해결: PIL을 사용한 형식 변환
from PIL import Image
import io
def convert_to_supported_format(
image_path: str,
target_format: str = "JPEG"
) -> bytes:
"""
지원되지 않는 이미지를 변환
Args:
image_path: 원본 이미지 경로
target_format: 대상 형식 (JPEG, PNG, WEBP)
Returns:
변환된 이미지 바이트
"""
with Image.open(image_path) as img:
# RGBA를 RGB로 변환 (JPEG는 알파 채널 미지원)
if img.mode == "RGBA" and target_format == "JPEG":
# 흰색 배경에 합성
background = Image.new("RGB", img.size, (255, 255, 255))
background.paste(img, mask=img.split()[3])
img = background
elif img.mode != "RGB":
img = img.convert("RGB")
# 메모리 내에서 변환
buffer = io.BytesIO()
img.save(buffer, format=target_format)
return buffer.getvalue()
사용 예시
def analyze_any_image(client, image_path, prompt):
"""모든 이미지 형식 지원"""
supported_formats = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
ext = Path(image_path).suffix.lower()
if ext not in supported_formats:
print(f"변환 필요: {ext} -> JPEG")
image_bytes = convert_to_supported_format(image_path, "JPEG")
base64_image = base64.b64encode(image_bytes).decode("utf-8")
image_url = f"data:image/jpeg;base64,{base64_image}"
else:
# 기존 로직
pass
3. 대용량 이미지 메모리 초과
# 오류 메시지 예시
"Image size exceeds maximum allowed (20MB)"
해결: 이미지 리사이징 및 최적화
from PIL import Image
import math
def resize_large_image(
image_path: str,
max_pixels: int = 4096 * 4096, # Claude 최대
quality: int = 85,
max_size_mb: float = 19.0
) -> bytes:
"""
대용량 이미지 최적화
Args:
image_path: 이미지 경로
max_pixels: 최대 픽셀 수
quality: JPEG 품질 (1-100)
max_size_mb: 최대 파일 크기 (MB)
Returns:
최적화된 이미지 바이트
"""
with Image.open(image_path) as img:
width, height = img.size
total_pixels = width * height
# 픽셀 수 최적화
if total_pixels > max_pixels:
ratio = math.sqrt(max_pixels / total_pixels)
new_size = (int(width * ratio), int(height * ratio))
img = img.resize(new_size, Image.LANCZOS)
# JPEG 압축
buffer = io.BytesIO()
if img.mode in ("RGBA", "P"):
img = img.convert("RGB")
img.save(
buffer,
format="JPEG",
quality=quality,
optimize=True
)
# 크기 체크
size_mb = len(buffer.getvalue()) / (1024 * 1024)
if size_mb > max_size_mb:
# 품질 낮추기
for q in range(quality - 5, 30, -5):
buffer = io.BytesIO()
img.save(buffer, format="JPEG", quality=q, optimize=True)
size_mb = len(buffer.getvalue()) / (1024 * 1024)
if size_mb <= max_size_mb:
break
return buffer.getvalue()
사용 예시
def analyze_large_image_safely(client, image_path, prompt):
"""대용량 이미지를 안전하게 분석"""
try:
# 먼저 시도
result = client.analyze_image(image_path, prompt)
return result
except Exception as e:
if "size exceeds" in str(e).lower():
print("대용량 이미지 감지. 최적화 후 재시도...")
# 최적화
optimized_bytes = resize_large_image(image_path)
base64