저는 HolySheep AI에서 2년 넘게 글로벌 개발자들의 API 통합을 지원해 온 엔지니어입니다. 하루에平均 50건 이상의 할당량 관련 문의를 받는데, 그 중 80%는 동일한 원인들 때문입니다. 이번 가이드에서는 AI API 자원 할당량의 핵심 개념부터 HolySheep AI 환경에서의 실전 최적화 전략까지, 실제 발생한 오류 시나리오와 함께 상세히 설명드리겠습니다.
가장 흔한 할당량 오류 4가지
프로덕션 환경에서 가장 자주 마주치는 오류들을 모았습니다. 이 오류들이 발생하는 근본 원인을 이해하면, 사전 예방이 가능합니다.
1. Rate Limit 초과 (429 Too Many Requests)
# HolySheep AI에서 Rate Limit 초과 시 반환되는 오류 응답
{
"error": {
"message": "Rate limit exceeded for model gpt-4.1.
Current: 1000 requests/min, Limit: 500 requests/min",
"type": "rate_limit_error",
"code": "rate_limit_exceeded",
"param": null,
"retry_after_ms": 2847
}
}
2. 토큰 할당량 소진 (401 Unauthorized)
# API 키가 유효하지만 잔여 할당량이 부족한 경우
{
"error": {
"message": "You have exceeded your monthly token allocation.
Please upgrade your plan or wait for reset (billing cycle: 1st of month)",
"type": "invalid_request_error",
"code": "quota_exceeded",
"param": null
}
}
3. TPM/RPM 초과 (Temporal Limit)
# 분당 토큰/요청 수 제한 초과
{
"error": {
"message": "Tokens per minute limit exceeded.
Current TPM: 150,000, Limit: 100,000.
Implement exponential backoff.",
"type": "rate_limit_error",
"code": "tpm_limit_exceeded"
}
}
4. 동시 요청 제한 초과
# 동시 연결 수 초과
{
"error": {
"message": "Concurrent request limit reached (max: 10).
Queue subsequent requests or implement connection pooling.",
"type": "rate_limit_error",
"code": "concurrent_limit_exceeded"
}
}
HolySheep AI 할당량 구조 이해하기
HolySheep AI는 개발자들이 효율적으로 API를 사용할 수 있도록 계층화된 할당량 체계를 제공합니다. 이 구조를 이해하면 불필요한 오류를 예방할 수 있습니다.
할당량 계층 구조
- 월간 토큰 할당량: 매달 reset되는 총 토큰 사용 한도
- TPM (Tokens Per Minute): 분당 처리 가능한 토큰 수
- RPM (Requests Per Minute): 분당 허용되는 API 호출 수
- RPD (Requests Per Day): 일일 요청 수 제한
- 동시 연결 제한: 동시에 허용되는 최대 연결 수
주요 모델별 할당량 비교
| 모델 | 가격 ($/MTok) | 기본 TPM | 기본 RPM |
|---|---|---|---|
| GPT-4.1 | $8.00 | 100,000 | 500 |
| Claude Sonnet 4 | $15.00 | 80,000 | 400 |
| Gemini 2.5 Flash | $2.50 | 150,000 | 1,000 |
| DeepSeek V3.2 | $0.42 | 120,000 | 600 |
실전 코드: 할당량 모니터링 및 관리
제가 실제 프로덕션에서 사용하는 할당량 모니터링 및 관리 코드를 공유합니다. 이 코드들은 HolySheep AI 환경에 최적화되어 있습니다.
1. 할당량 모니터링 데코레이터
import time
import logging
from functools import wraps
from typing import Callable, Dict, Any
import requests
logger = logging.getLogger(__name__)
class HolySheepAIClient:
"""HolySheep AI API 클라이언트 - 할당량 모니터링 포함"""
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"
}
# 로컬 할당량 트래커
self.quota_tracker = {
"requests_this_minute": 0,
"tokens_this_minute": 0,
"last_reset": time.time(),
"minute_window": 60
}
def get_usage_stats(self) -> Dict[str, Any]:
"""현재 사용량 통계 조회"""
try:
response = requests.get(
f"{self.base_url}/usage/current",
headers=self.headers,
timeout=10
)
if response.status_code == 200:
return response.json()
else:
return {"error": f"Status {response.status_code}"}
except Exception as e:
logger.error(f"사용량 조회 실패: {e}")
return {"error": str(e)}
def reset_minute_tracker(self):
"""분당 트래커 초기화"""
current_time = time.time()
if current_time - self.quota_tracker["last_reset"] >= self.quota_tracker["minute_window"]:
self.quota_tracker["requests_this_minute"] = 0
self.quota_tracker["tokens_this_minute"] = 0
self.quota_tracker["last_reset"] = current_time
def check_local_limit(self, estimated_tokens: int, rpm_limit: int = 500) -> bool:
"""로컬에서 제한 체크 (API 호출 전)"""
self.reset_minute_tracker()
if self.quota_tracker["requests_this_minute"] >= rpm_limit:
logger.warning(f"RPM 제한 도달: {rpm_limit}")
return False
if self.quota_tracker["tokens_this_minute"] + estimated_tokens > 100000:
logger.warning(f"TPM 제한 초과 예상")
return False
return True
def tracked_completion(self, messages: list, model: str = "gpt-4.1") -> Dict:
"""사용량 추적 기능이 포함된 채팅 완료 API"""
self.reset_minute_tracker()
# 대략적인 토큰 수估算 (실제 사용량은 응답에서 확인)
estimated_input_tokens = sum(len(str(m)) for m in messages) // 4
if not self.check_local_limit(estimated_input_tokens):
raise Exception("429: Rate limit approaching - implement backoff")
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": model,
"messages": messages,
"max_tokens": 2048
},
timeout=60
)
self.quota_tracker["requests_this_minute"] += 1
if response.status_code == 200:
result = response.json()
usage = result.get("usage", {})
self.quota_tracker["tokens_this_minute"] += (
usage.get("prompt_tokens", 0) +
usage.get("completion_tokens", 0)
)
return result
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
raise Exception(f"429: Rate limit exceeded. Retry after {retry_after}s")
else:
raise Exception(f"API Error {response.status_code}: {response.text}")
except requests.exceptions.Timeout:
raise Exception("408: Request timeout")
except requests.exceptions.ConnectionError:
raise Exception("503: Connection error - service unavailable")
def with_retry_and_backoff(max_retries: int = 3, base_delay: float = 1.0):
"""지수 백오프와 재시도 데코레이터"""
def decorator(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if "429" in str(e) and attempt < max_retries - 1:
delay = base_delay * (2 ** attempt)
logger.info(f"재시도 {attempt + 1}/{max_retries}, {delay}s 대기")
time.sleep(delay)
else:
raise
raise Exception("Maximum retries exceeded")
return wrapper
return decorator
사용 예시
if __name__ == "__main__":
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# 현재 사용량 확인
stats = client.get_usage_stats()
print(f"현재 사용량: {stats}")
# 할당량 추적しながら API 호출
try:
@with_retry_and_backoff(max_retries=3)
def call_ai():
return client.tracked_completion(
messages=[{"role": "user", "content": "Hello!"}],
model="gpt-4.1"
)
result = call_ai()
print(f"응답: {result['choices'][0]['message']['content']}")
except Exception as e:
print(f"오류 발생: {e}")
2. 일괄 처리 및 할당량 최적화
import asyncio
import aiohttp
from collections import deque
import time
class RateLimitedAsyncClient:
"""비동기 AI API 클라이언트 - Rate Limit 자동 처리"""
def __init__(
self,
api_key: str,
rpm_limit: int = 500,
tpm_limit: int = 100000,
requests_per_second: float = 8.0 # RPM의 안전한 비율
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
self.min_request_interval = 1.0 / requests_per_second
# 토큰 버킷 알고리즘
self.token_bucket = {
"tokens": tpm_limit,
"max_tokens": tpm_limit,
"refill_rate": tpm_limit / 60 # 분당 refill
}
self.request_queue = deque()
self.processing = False
def _estimate_tokens(self, messages: list) -> int:
"""토큰 수概算 (단순估算)"""
return sum(len(str(m)) // 4 for m in messages)
async def _wait_for_token_bucket(self, tokens_needed: int):
"""토큰 버킷이 채워지기를 대기"""
while self.token_bucket["tokens"] < tokens_needed:
# 1초 대기 후 토큰 refill
await asyncio.sleep(1)
elapsed = 1
self.token_bucket["tokens"] = min(
self.token_bucket["max_tokens"],
self.token_bucket["tokens"] + self.token_bucket["refill_rate"] * elapsed
)
async def _wait_for_rate_limit(self):
"""분당 요청 수 제한 대기"""
now = time.time()
if hasattr(self, '_last_request_time'):
elapsed = now - self._last_request_time
if elapsed < self.min_request_interval:
await asyncio.sleep(self.min_request_interval - elapsed)
self._last_request_time = time.time()
async def chat_completion_async(
self,
messages: list,
model: str = "gpt-4.1",
max_retries: int = 3
) -> dict:
"""비동기 채팅 완료 API 호출"""
estimated_tokens = self._estimate_tokens(messages)
for attempt in range(max_retries):
try:
# Rate Limit 체크
await self._wait_for_rate_limit()
await self._wait_for_token_bucket(estimated_tokens)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 200:
result = await response.json()
# 사용량 업데이트
usage = result.get("usage", {})
consumed = usage.get("total_tokens", estimated_tokens)
self.token_bucket["tokens"] -= consumed
return result
elif response.status == 429:
retry_after = int(response.headers.get("Retry-After", 5))
print(f"Rate limit - {retry_after}s 대기 (시도 {attempt + 1})")
await asyncio.sleep(retry_after)
continue
elif response.status == 401:
raise Exception("401: Invalid API key or quota exceeded")
else:
text = await response.text()
raise Exception(f"{response.status}: {text}")
except asyncio.TimeoutError:
print(f"Timeout (시도 {attempt + 1}/{max_retries})")
if attempt == max_retries - 1:
raise Exception("408: Request timeout after retries")
except aiohttp.ClientError as e:
print(f"Connection error: {e}")
await asyncio.sleep(2 ** attempt)
raise Exception("Max retries exceeded")
async def batch_process(
self,
batch_requests: list,
model: str = "gpt-4.1",
concurrency: int = 5
):
"""배치 처리 - 동시성 제어 포함"""
semaphore = asyncio.Semaphore(concurrency)
async def process_single(req_id: int, messages: list):
async with semaphore:
try:
result = await self.chat_completion_async(messages, model)
return {"id": req_id, "status": "success", "result": result}
except Exception as e:
return {"id": req_id, "status": "error", "error": str(e)}
tasks = [
process_single(i, req)
for i, req in enumerate(batch_requests)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
사용 예시
async def main():
client = RateLimitedAsyncClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
rpm_limit=500,
tpm_limit=100000,
requests_per_second=8.0
)
# 일괄 요청
batch = [
[{"role": "user", "content": f"요청 {i}: 설명해줘"}]
for i in range(20)
]
start_time = time.time()
results = await client.batch_process(batch, concurrency=5)
elapsed = time.time() - start_time
success = sum(1 for r in results if isinstance(r, dict) and r["status"] == "success")
print(f"완료: {success}/{len(batch)} 성공, 소요시간: {elapsed:.2f}초")
if __name__ == "__main__":
asyncio.run(main())
3. 할당량 초과 감지 및 알림 시스템
import smtplib
from dataclasses import dataclass
from typing import Optional
import logging
@dataclass
class QuotaAlert:
threshold_percent: float = 80.0 # 80% 이상 시 알림
critical_percent: float = 95.0 # 95% 이상 시 긴급
@dataclass
class QuotaStatus:
used_tokens: int
total_quota: int
remaining_tokens: int
reset_date: str
usage_percent: float
class QuotaAlertManager:
"""할당량 알림 관리자"""
def __init__(self, api_key: str, alert: QuotaAlert = None):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.alert = alert or QuotaAlert()
self.logger = logging.getLogger(__name__)
def check_quota_status(self) -> QuotaStatus:
"""현재 할당량 상태 확인"""
import requests
try:
response = requests.get(
f"{self.base_url}/quota/status",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=10
)
if response.status_code == 200:
data = response.json()
return QuotaStatus(
used_tokens=data["used_tokens"],
total_quota=data["total_quota"],
remaining_tokens=data["remaining"],
reset_date=data["next_reset"],
usage_percent=(data["used_tokens"] / data["total_quota"]) * 100
)
else:
self.logger.error(f"할당량 조회 실패: {response.status_code}")
return None
except Exception as e:
self.logger.error(f"할당량 확인 중 오류: {e}")
return None
def should_alert(self, status: QuotaStatus) -> Optional[str]:
"""알림 필요 여부 판단"""
if status.usage_percent >= self.alert.critical_percent:
return "CRITICAL"
elif status.usage_percent >= self.alert.threshold_percent:
return "WARNING"
return None
def send_alert(self, alert_type: str, status: QuotaStatus):
"""알림 전송"""
message = f"""
HolySheep AI 할당량 알림
========================
레벨: {alert_type}
사용률: {status.usage_percent:.1f}%
사용량: {status.used_tokens:,} / {status.total_quota:,}
잔여: {status.remaining_tokens:,}
리셋일: {status.reset_date}
"""
self.logger.warning(message)
# 실제로는 이메일/Slack/Discord 등으로 전송
# self._send_email(message)
# self._send_slack(message)
return message
def monitor_loop(self, check_interval: int = 300):
"""모니터링 루프 (5분마다 체크)"""
import time
while True:
status = self.check_quota_status()
if status:
alert_type = self.should_alert(status)
if alert_type:
self.send_alert(alert_type, status)
time.sleep(check_interval)
사용 예시
if __name__ == "__main__":
manager = QuotaAlertManager(
api_key="YOUR_HOLYSHEEP_API_KEY",
alert=QuotaAlert(threshold_percent=80.0, critical_percent=95.0)
)
status = manager.check_quota_status()
if status:
print(f"현재 사용률: {status.usage_percent:.1f}%")
print(f"잔여 토큰: {status.remaining_tokens:,}")
alert_type = manager.should_alert(status)
if alert_type:
manager.send_alert(alert_type, status)
HolySheep AI 환경에서의 최적 할당량 활용
HolySheep AI를 활용하면 다른 플랫폼 대비 더 효율적으로 할당량을 관리할 수 있습니다. 제가 직접 테스트하고 검증한 최적화 팁을 공유합니다.
모델 선택에 따른 할당량 절약
# 비용 최적화를 위한 모델 선택 가이드
"""
상황별 최적 모델 추천 (HolySheep AI 기준)
1. 대량 배치 처리 (비용 최우선)
→ DeepSeek V3.2: $0.42/MTok (가장 저렴)
→ Gemini 2.5 Flash: $2.50/MTok (가격 대비 성능 우수)
2. 대화형 인터페이스 (품질 + 속도 균형)
→ Gemini 2.5 Flash: 응답速度快, 비용适中
→ Claude Sonnet 4: $15/MTok (고품질 필요시)
3. 복잡한 추론 작업 (고품질)
→ GPT-4.1: $8/MTok (다양한 작업 대응)
→ Claude Sonnet 4: 긴 컨텍스트 처리 우수
"""
HolySheep AI에서 비용 최적화 호출 예시
def cost_optimized_completion(client, prompt: str, task_type: str) -> dict:
"""작업 유형에 따른 최적 모델 선택"""
model_mapping = {
"simple_qa": {
"model": "deepseek-v3.2",
"max_tokens": 500,
"temperature": 0.3
},
"creative": {
"model": "gpt-4.1",
"max_tokens": 1500,
"temperature": 0.8
},
"code_generation": {
"model": "claude-sonnet-4",
"max_tokens": 2000,
"temperature": 0.2
},
"fast_summary": {
"model": "gemini-2.5-flash",
"max_tokens": 800,
"temperature": 0.1
}
}
config = model_mapping.get(task_type, model_mapping["simple_qa"])
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {client.api_key}"},
json={
"model": config["model"],
"messages": [{"role": "user", "content": prompt}],
"max_tokens": config["max_tokens"],
"temperature": config["temperature"]
}
)
return response.json()
비용 비교 테스트
def compare_costs(task: str, content_length: int) -> dict:
"""동일 작업 각 모델 비용 비교"""
# 약 1000 토큰 입력 가정
input_tokens = content_length // 4
models = {
"GPT-4.1": {"price_per_mtok": 8.00, "output_ratio": 0.3},
"Claude Sonnet 4": {"price_per_mtok": 15.00, "output_ratio": 0.25},
"Gemini 2.5 Flash": {"price_per_mtok": 2.50, "output_ratio": 0.3},
"DeepSeek V3.2": {"price_per_mtok": 0.42, "output_ratio": 0.25}
}
results = {}
for name, info in models.items():
output_tokens = int(input_tokens * info["output_ratio"])
total_tokens = input_tokens + output_tokens
cost = (total_tokens / 1_000_000) * info["price_per_mtok"]
results[name] = {
"total_tokens": total_tokens,
"estimated_cost_usd": round(cost, 4)
}
return results
실제 비용 절감 사례
| 시나리오 | 변경 전 | 변경 후 | 절감률 |
|---|---|---|---|
| 일일 10만 요청 QA 봇 | GPT-4.1 only | Simple QA → DeepSeek V3.2 | 약 95% |
| 문서 요약 서비스 | Claude Sonnet 4 | Gemini 2.5 Flash | 약 83% |
| 배치 데이터 처리 | TPM 100K 소진 | 배치 스케줄링 + 모델 혼합 | 2배 처리량 증가 |
자주 발생하는 오류와 해결책
오류 1: "429: Rate limit exceeded" - 무한 재시도 루프
# ❌ 잘못된 접근: 재시도 없이 무한 대기
def bad_example():
while True:
try:
response = call_api()
return response
except Exception as e:
print(f"오류: {e}")
# 무한 대기 - CPU 낭비, 할당량 회복 지연
✅ 올바른 접근: 지수 백오프와 최대 재시도 횟수 제한
def correct_retry_with_backoff(api_call_func, max_retries=5):
"""
지수 백오프(Exponential Backoff) 구현
HolySheep AI 권장: base_delay=1초, max_delay=32초
"""
base_delay = 1.0
max_delay = 32.0
for attempt in range(max_retries):
try:
return api_call_func()
except Exception as e:
if "429" in str(e):
delay = min(base_delay * (2 ** attempt), max_delay)
# 지터(jitter) 추가: 동시 요청者们 분산
import random
delay += random.uniform(0, 0.5)
print(f"Rate limit 도달. {delay:.1f}초 후 재시도 ({attempt + 1}/{max_retries})")
time.sleep(delay)
else:
# 429 외 오류는 즉시 실패
raise
raise Exception("최대 재시도 횟수 초과 - 관리자에게 문의하세요")
오류 2: "401: Invalid API key" - 할당량 초과误会
# ❌ 잘못된 접근: 키 유효성 확인 안함
def bad_auth():
headers = {"Authorization": f"Bearer {api_key}"}
# 토큰만 확인하고 실제 할당량 확인 안함
✅ 올바른 접근: 사전 검증 및 명확한 에러 구분
def robust_auth_check(client):
"""
401 오류의 실제 원인 파악
1. API 키 오타/형식 오류
2. 할당량 소진
3. 계정 정지/만료
"""
try:
# 먼저 간단한 API 호출로 테스트
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {client.api_key}"},
timeout=5
)
if response.status_code == 200:
print("✅ API 키 유효")
return True
elif response.status_code == 401:
# 401의 실제 원인 파악
error_detail = response.json()
if "quota" in str(error_detail).lower():
print("❌ 할당량 소진 - 요금제 업그레이드 필요")
print(f" 다음 리셋일: {get_reset_date()}")
elif "invalid" in str(error_detail).lower():
print("❌ API 키 오류 - 키를 확인하세요")
else:
print(f"❌ 인증 오류: {error_detail}")
return False
else:
print(f"⚠️ 예상치 못한 응답: {response.status_code}")
return False
except Exception as e:
print(f"❌ 연결 실패: {e}")
return False
HolySheep AI에서 할당량 상태 확인
def get_holysheep_quota_status(api_key: str):
"""HolySheep AI 할당량 상태 확인"""
try:
response = requests.get(
"https://api.holysheep.ai/v1/quota/status",
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
if response.status_code == 200:
data = response.json()
print(f"월간 사용량: {data['used_tokens']:,} / {data['total_quota']:,}")
print(f"잔여: {data['remaining']:,} 토큰")
print(f"다음 리셋: {data['next_reset']}")
return data
return None
except Exception as e:
print(f"할당량 조회 실패: {e}")
return None
오류 3: 할당량 초과로 인한 서비스 중단
# ❌ 잘못된 접근: 사전 체크 없이 최대 부하로 요청
def bad_approach():
tasks = [create_task(i) for i in range(1000)]
results = []
for task in tasks:
results.append(call_api(task)) # 한 번에 1000개 전송
return results
✅ 올바른 접근: 할당량感知 요청 처리
class QuotaAwareRequestHandler:
"""
HolySheep AI 할당량을感知하여 자동으로 요청 조절
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.daily_limit = 1_000_000 # 일일 토큰 제한 (설정 가능)
self.used_today = 0
self.last_check = datetime.date.today()
def _check_daily_reset(self):
"""자마다 일일 카운터 초기화"""
today = datetime.date.today()
if today > self.last_check:
self.used_today = 0
self.last_check = today
def _estimate_request_cost(self, messages: list, max_tokens: int = 1000) -> int:
"""요청 비용概算"""
input_tokens = sum(len(str(m)) // 4 for m in messages)
return input_tokens + max_tokens
def can_make_request(self, messages: list, max_tokens: int = 1000) -> tuple:
"""
요청 가능 여부 확인
Returns: (can_proceed: bool, reason: str, remaining: int)
"""
self._check_daily_reset()
estimated_cost = self._estimate_request_cost(messages, max_tokens)
remaining = self.daily_limit - self.used_today
if estimated_cost > remaining:
# 할당량 초과 예상 - 요청 스케줄링 권장
return (
False,
f"일일 할당량 부족. 필요: {estimated_cost:,}, 잔여: {remaining:,}",
remaining
)
if remaining < self.daily_limit * 0.1:
# 10% 미만 잔여 - 경고
return (
True,
f"⚠️ 할당량 부족 경고! 잔여 {remaining:,} 토큰",
remaining
)
return (True, "OK", remaining)
def execute_with_quota_check(self, messages: list) -> dict:
"""할당량 체크 후 실행"""
can_proceed, reason, remaining = self.can_make_request(messages)
if not can_proceed:
return {
"success": False,
"error": "quota_exceeded",
"message": reason,
"suggestion": "30분 후 재시도하거나 다음 날에 요청하세요"
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "gpt-4.1",
"messages": messages,
"max_tokens": 500
},
timeout=60
)
if response.status_code == 200:
result = response.json()
usage = result.get("usage", {})
tokens_used = usage.get("total_tokens", 0)
self.used_today += tokens_used
return {
"success": True,
"result": result,
"tokens_used": tokens_used,
"remaining_today": self.daily_limit - self.used_today
}
elif response.status_code == 429:
return {
"success": False,
"error": "rate_limit",
"message": "Rate limit 초과. 잠시 후 재시도하세요",
"retry_after": response.headers.get("Retry-After", 30)
}
else:
return {
"success": False,
"error": "api_error",
"message": f"API 오류: {response.status_code}"
}
except Exception as e:
return {
"success": False,
"error": "exception",
"message": str(e)
}
배치 처리에서의 활용
def process_large_batch(requests: list, api_key: str):
"""대량 요청을 할당량 범위 내에서 분할 처리"""
handler = QuotaAwareRequestHandler(api_key)
results = []
batch_size = 50 # 배치 크기
pause_after_batches = 5 # 이 수만큼 배치 후 일시 정지
for i in range(0, len(requests), batch_size):
batch = requests[i:i + batch_size]
batch_num = i // batch_size + 1
print(f"\n배치 {batch_num} 처리 중...")
for req in batch:
result = handler.execute_with_quota_check(req)
if result["success"]:
results.append(result["result"])
else:
if "quota" in result.get("error", ""):
# 할당량 초과 - 일시 정지 후 계속
print(f"⚠️ 할당량 경고: {result['message']}")
print("60초 대기 후 계속...")
time.sleep(60)
# 재시도
result = handler.execute_with_quota_check(req)
if result["success"]:
results.append(result["result"])
else:
print(f"오류: {result['message']}")
# 배치 간 일시 정지 (Rate Limit 방지)
if batch_num % pause_after_batches == 0:
print(f"{batch_num} 배치 완료. 30초 대기...")
time.sleep(30)
return results
오류 4: 동시 요청으로 인한 연결 풀 소진
# ❌ 잘못된 접근: 연결 풀 미관리
def bad_concurrent_requests():
tasks = []
for i in range(100):
# 각각 새 연결 생성
tasks.append(requests.post(url, json=data)) # 연결 풀 고갈!
return tasks
✅ 올바른 접근: 연결 풀링 및 동시성 제어
import concurrent.futures
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_pooling():
"""연결 풀링이 구성된 세션 생성"""
session = requests.Session()
# 연결 풀 설정
adapter = HTTPAdapter(
pool_connections=10, # 풀의 연결 수
pool_maxsize=20, # 풀