생성형 AI 서비스를 운영하다 보면 가장 흔하게 마주치는 문제가 바로 Rate Limit 초과입니다. 하루平均 50만 건의 API 호출을 처리하는 production 환경에서 429 Too Many Requests 에러가 발생하면 서비스 전체가 중단될 수 있습니다.
이 튜토리얼에서는 HolySheep AI 게이트웨이를 활용해 Rate Limit 발생 시 자동 failover, 회로 차단기 패턴, 모델 라우팅을 구현하는 방법을 상세히 설명합니다.
1. Rate Limit 초과 에러 시나리오 분석
실제 production 환경에서 발생했던 에러 로그를 먼저 살펴보겠습니다:
# 에러 시나리오 1: GPT-4.1 Rate Limit 초과
HTTP 429 Too Many Requests
Response Body: {
"error": {
"message": "Rate limit reached for gpt-4.1 in organization org-xxxxx
on requests per day. Please retry after 86,400 seconds",
"type": "rate_limit_exceeded",
"code": "daily_rate_limit_exceeded"
}
}
에러 시나리오 2: 토큰 Rate Limit 초과
HTTP 429 Too Many Requests
{
"error": {
"message": "Rate limit reached for gpt-4.1 in organization org-xxxxx
on requests per min. Please retry after 58 seconds",
"type": "rate_limit_exceeded",
"code": "minute_limit_exceeded",
"param": null,
"rate": {
"limit_tokens": 150000,
"used_tokens": 150000,
"reset_at": "2026-04-29T07:00:00Z"
}
}
}
에러 시나리오 3: TPM(Time-Per-Minute) 초과
ConnectionError: timeout after 30s
httpx.ConnectTimeout: Connection timeout during request
에러 시나리오 4: 컨텍스트 창 초과 (가장 흔한 400 에러)
HTTP 400 Bad Request
{
"error": {
"message": "Maximum context window exceeded.
Maximum: 128000 tokens, Requested: 156,234 tokens",
"type": "invalid_request_error",
"code": "context_length_exceeded"
}
}위 4가지 에러 유형을 처리하기 위해 4단계 폴백 전략을 구현하겠습니다.
2. 핵심 아키텍처: Hierarchical Fallback System
# holy_sheep_gateway.py
"""
HolySheep AI 게이트웨이 - Rate Limit 대응 + 다중 모델 라우팅
Production-ready 구현체
"""
import asyncio
import time
import logging
from typing import Optional, Dict, List, Any
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import httpx
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelTier(Enum):
"""모델 계층 구조 - 비용 순서로 정렬"""
TIER_1_PREMIUM = "gpt-4.1" # $8/MTok
TIER_2_STANDARD = "claude-sonnet-4-5" # $15/MTok
TIER_3_BUDGET = "gemini-2.5-flash" # $2.50/MTok
TIER_4_MINIMAL = "deepseek-v3.2" # $0.42/MTok
class CircuitState(Enum):
"""회로 차단기 상태"""
CLOSED = "closed" # 정상 - 요청 허용
OPEN = "open" # 차단 - 요청 거부
HALF_OPEN = "half_open" # 반개방 - 테스트 허용
@dataclass
class RateLimitConfig:
"""Rate Limit 설정"""
rpm_limit: int = 500 # 요청/분
tpm_limit: int = 150000 # 토큰/분
daily_limit: int = 10000000 # 일일 제한
retry_after_seconds: int = 60
@dataclass
class CircuitBreaker:
"""회로 차단기 구현"""
failure_threshold: int = 5 # N회 실패 시 차단
success_threshold: int = 3 # 반개방 후 N회 성공 시 복구
timeout_seconds: int = 30 # 반개방 전환 대기 시간
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
success_count: int = 0
last_failure_time: float = 0
recovery_attempts: Dict[str, int] = field(default_factory=dict)
class HolySheepGateway:
"""
HolySheep AI 게이트웨이 클라이언트
주요 기능:
1. 자동 Rate Limit 감지 및 failover
2. 회로 차단기 패턴 적용
3. 4단계 모델 폴백
4. 실시간 비용 추적
"""
BASE_URL = "https://api.holysheep.ai/v1"
# 모델별 지연 시간 임계치 (ms) - SLA 기준
LATENCY_THRESHOLDS = {
"gpt-4.1": 3000,
"claude-sonnet-4-5": 4000,
"gemini-2.5-flash": 1500,
"deepseek-v3.2": 2000
}
# 토큰 가격 ($/MTok) - HolySheep 공식 가격
TOKEN_PRICES = {
"gpt-4.1": 8.0,
"claude-sonnet-4-5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
def __init__(self, api_key: str):
self.api_key = api_key
self.rate_limit_config = RateLimitConfig()
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
self.request_counts: Dict[str, List[float]] = defaultdict(list)
self.cost_tracker: Dict[str, float] = defaultdict(float)
# 각 모델에 대한 회로 차단기 초기화
for tier in ModelTier:
self.circuit_breakers[tier.value] = CircuitBreaker()
async def chat_completion(
self,
messages: List[Dict],
primary_model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> Dict[str, Any]:
"""
메인 API 호출 - 자동 폴백 포함
Args:
messages: OpenAI 형식 메시지
primary_model: 기본 모델
temperature: 창의성 온도
max_tokens: 최대 출력 토큰
Returns:
API 응답 딕셔너리
"""
# 1단계: 기본 모델로 시도
model_sequence = self._get_fallback_sequence(primary_model)
last_error = None
for attempt, model in enumerate(model_sequence):
try:
logger.info(f"[Attempt {attempt + 1}] Using model: {model}")
# 회로 차단기 체크
if not self._is_circuit_allowed(model):
logger.warning(f"Circuit breaker OPEN for {model}, skipping...")
continue
# Rate Limit 체크
if self._is_rate_limited(model):
logger.warning(f"Rate limit detected for {model}, skipping...")
continue
response = await self._make_request(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
# 성공 - 회로 차단기 복구 시도
self._on_success(model)
self._track_cost(model, response)
return {
"model": response.get("model", model),
"content": response["choices"][0]["message"]["content"],
"usage": response.get("usage", {}),
"fallback_level": attempt,
"latency_ms": response.get("latency_ms", 0)
}
except RateLimitError as e:
logger.error(f"Rate limit error with {model}: {e}")
self._on_failure(model)
last_error = e
continue
except ContextLengthError as e:
# 컨텍스트 초과 - 모델을 바꾸고 프롬프트를 압축
logger.error(f"Context length error: {e}")
messages = self._truncate_messages(messages, model)
continue
except AuthenticationError as e:
logger.error(f"Auth error - check API key: {e}")
raise
except Exception as e:
logger.error(f"Unexpected error with {model}: {e}")
self._on_failure(model)
last_error = e
continue
raise AllModelsFailedError(f"All models failed. Last error: {last_error}")
async def _make_request(
self,
model: str,
messages: List[Dict],
temperature: float,
max_tokens: int,
**kwargs
) -> Dict[str, Any]:
"""실제 API 요청 수행"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
start_time = time.time()
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 429:
retry_after = response.headers.get("Retry-After", 60)
raise RateLimitError(f"Rate limited, retry after {retry_after}s")
if response.status_code == 400:
error_data = response.json()
if "context_length_exceeded" in str(error_data):
raise ContextLengthError("Context window exceeded")
if response.status_code == 401:
raise AuthenticationError("Invalid API key")
if response.status_code != 200:
raise APIError(f"HTTP {response.status_code}: {response.text}")
result = response.json()
result["latency_ms"] = latency_ms
return result
def _get_fallback_sequence(self, primary_model: str) -> List[str]:
"""폴백 순서 결정 - 비용 효율성 기반"""
model_tiers = {
"gpt-4.1": 0,
"claude-sonnet-4-5": 1,
"gemini-2.5-flash": 2,
"deepseek-v3.2": 3
}
primary_index = model_tiers.get(primary_model, 0)
# 기본 모델보다 뒤의 모델만 폴백으로 사용
return [t for t, idx in sorted(model_tiers.items(), key=lambda x: x[1])
if idx >= primary_index]
def _is_circuit_allowed(self, model: str) -> bool:
"""회로 차단기 상태 확인"""
breaker = self.circuit_breakers.get(model)
if not breaker:
return True
if breaker.state == CircuitState.CLOSED:
return True
if breaker.state == CircuitState.OPEN:
# 타임아웃 확인
if time.time() - breaker.last_failure_time > breaker.timeout_seconds:
breaker.state = CircuitState.HALF_OPEN
logger.info(f"Circuit breaker HALF-OPEN for {model}")
return True
return False
# HALF_OPEN - 1개만 허용
return True
def _on_success(self, model: str):
"""성공 시 회로 차단기 복구"""
breaker = self.circuit_breakers.get(model)
if not breaker:
return
if breaker.state == CircuitState.HALF_OPEN:
breaker.success_count += 1
if breaker.success_count >= breaker.success_threshold:
breaker.state = CircuitState.CLOSED
breaker.failure_count = 0
breaker.success_count = 0
logger.info(f"Circuit breaker CLOSED for {model}")
elif breaker.state == CircuitState.CLOSED:
breaker.failure_count = max(0, breaker.failure_count - 1)
def _on_failure(self, model: str):
"""실패 시 회로 차단기 차단"""
breaker = self.circuit_breakers.get(model)
if not breaker:
return
breaker.failure_count += 1
breaker.last_failure_time = time.time()
if breaker.failure_count >= breaker.failure_threshold:
breaker.state = CircuitState.OPEN
breaker.success_count = 0
logger.warning(f"Circuit breaker OPENED for {model} after {breaker.failure_count} failures")
def _is_rate_limited(self, model: str) -> bool:
"""Rate Limit 상태 확인"""
current_time = time.time()
cutoff_time = current_time - 60 # 1분 전
# 1분 내 요청 수
recent_requests = [
t for t in self.request_counts[model]
if t > cutoff_time
]
if len(recent_requests) >= self.rate_limit_config.rpm_limit:
return True
self.request_counts[model] = recent_requests
self.request_counts[model].append(current_time)
return False
def _track_cost(self, model: str, response: Dict):
"""비용 추적"""
usage = response.get("usage", {})
total_tokens = usage.get("total_tokens", 0)
cost = (total_tokens / 1_000_000) * self.TOKEN_PRICES.get(model, 0)
self.cost_tracker[model] += cost
def _truncate_messages(self, messages: List[Dict], model: str) -> List[Dict]:
"""메시지 트렁케이션 - 컨텍스트 초과 방지"""
# 시스템 프롬프트 유지
system_msg = messages[0] if messages and messages[0]["role"] == "system" else None
# 최근 대화 유지 (대략 절반)
remaining = messages[1:][-10:] if len(messages) > 1 else []
result = []
if system_msg:
result.append(system_msg)
result.append({
"role": "system",
"content": f"[이전 대화 요약] {len(messages) - 1}개의 메시지가 요약되었습니다."
})
result.extend(remaining)
return result
def get_cost_summary(self) -> Dict[str, float]:
"""비용 요약 반환"""
total_cost = sum(self.cost_tracker.values())
return {
"by_model": dict(self.cost_tracker),
"total": total_cost,
"currency": "USD"
}
커스텀 예외 클래스
class RateLimitError(Exception):
pass
class ContextLengthError(Exception):
pass
class AuthenticationError(Exception):
pass
class APIError(Exception):
pass
class AllModelsFailedError(Exception):
pass3. Rate Limit 감지 및 자동 Retry 로직
# retry_handler.py
"""
지수 백오프 + Jitter가 적용된 Retry 로직
Rate Limit 최적화
"""
import asyncio
import random
from typing import Callable, Any, Optional
from functools import wraps
import time
class RetryHandler:
"""
지수 백오프(Exponential Backoff) + Jitter 적용 Retry 핸들러
Rate Limit Retry 시나리오:
- 1차: 1초 후 재시도
- 2차: 2초 후 재시도
- 3차: 4초 후 재시도
- 4차: 8초 후 재시도
- 5차: 16초 후 재시도
Jitter 추가: 동시 요청 충돌 방지
"""
def __init__(
self,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
def calculate_delay(self, attempt: int, retry_after: Optional[int] = None) -> float:
"""지연 시간 계산"""
# Rate Limit 헤더에 명시된 시간 우선 적용
if retry_after:
return min(retry_after, self.max_delay)
# 지수 백오프 계산
delay = self.base_delay * (self.exponential_base ** attempt)
delay = min(delay, self.max_delay)
# Jitter 추가 (0.5 ~ 1.5 배)
if self.jitter:
delay *= (0.5 + random.random())
return delay
async def execute_with_retry(
self,
func: Callable,
*args,
**kwargs
) -> Any:
"""재시도 로직 실행"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
if asyncio.iscoroutinefunction(func):
return await func(*args, **kwargs)
else:
return func(*args, **kwargs)
except RateLimitError as e:
last_exception = e
if attempt == self.max_retries:
break
# Retry-After 헤더 파싱
retry_after = self._parse_retry_after(str(e))
delay = self.calculate_delay(attempt, retry_after)
print(f"⏳ Attempt {attempt + 1}/{self.max_retries + 1} failed. "
f"Retrying in {delay:.2f}s...")
await asyncio.sleep(delay)
except ContextLengthError as e:
# 컨텍스트 에러는 재시도로 해결 불가
raise
except Exception as e:
last_exception = e
if attempt == self.max_retries:
break
delay = self.calculate_delay(attempt)
print(f"❌ Attempt {attempt + 1}/{self.max_retries + 1} failed: {e}. "
f"Retrying in {delay:.2f}s...")
await asyncio.sleep(delay)
raise AllModelsFailedError(f"All retries exhausted. Last error: {last_exception}")
def _parse_retry_after(self, error_message: str) -> Optional[int]:
"""에러 메시지에서 Retry-After 시간 파싱"""
import re
match = re.search(r"retry after (\d+)", error_message.lower())
if match:
return int(match.group(1))
return None
데코레이터 버전
def with_retry(
max_retries: int = 5,
base_delay: float = 1.0,
jitter: bool = True
):
"""Retry 데코레이터"""
handler = RetryHandler(
max_retries=max_retries,
base_delay=base_delay,
jitter=jitter
)
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
return await handler.execute_with_retry(func, *args, **kwargs)
return wrapper
return decorator
사용 예시
async def example_usage():
"""HolySheep AI 게이트웨이 사용 예시"""
# API 키 설정 (HolySheep 가입 후 발급)
api_key = "YOUR_HOLYSHEEP_API_KEY"
gateway = HolySheepGateway(api_key)
retry_handler = RetryHandler(max_retries=3)
messages = [
{"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
{"role": "user", "content": "안녕하세요! 최근 기술 트렌드에 대해 설명해 주세요."}
]
try:
# 자동 폴백으로 API 호출
result = await retry_handler.execute_with_retry(
gateway.chat_completion,
messages=messages,
primary_model="gpt-4.1", # 기본 모델
temperature=0.7,
max_tokens=1500
)
print(f"✅ Success! Model: {result['model']}")
print(f"📊 Fallback Level: {result['fallback_level']}")
print(f"⏱️ Latency: {result['latency_ms']:.0f}ms")
print(f"💬 Response: {result['content'][:200]}...")
# 비용 확인
cost_summary = gateway.get_cost_summary()
print(f"💰 Total Cost: ${cost_summary['total']:.4f}")
except AllModelsFailedError as e:
print(f"🚨 All models failed: {e}")
if __name__ == "__main__":
asyncio.run(example_usage())4. 모델별 Cost-Efficiency 비교표
HolySheep AI에서 제공하는 주요 모델의 가격 및 성능 특성을 비교합니다:
| 모델 | 가격 ($/MTok) | 컨텍스트 창 | 평균 지연시간 | 적합 용도 | Rate Limit |
|---|---|---|---|---|---|
| GPT-4.1 | $8.00 | 128K 토큰 | 2,800ms | 고급 추론, 복잡한 코드 | 500 RPM / 150K TPM |
| Claude Sonnet 4.5 | $15.00 | 200K 토큰 | 3,200ms | 장문 분석, 컨텍스트 활용 | 400 RPM / 200K TPM |
| Gemini 2.5 Flash | $2.50 | 1M 토큰 | 1,200ms | 빠른 응답, 대량 처리 | 1,000 RPM / 1M TPM |
| DeepSeek V3.2 | $0.42 | 640K 토큰 | 1,800ms | 비용 최적화, Bulk 처리 | 2,000 RPM / 10M TPM |
비용 절감 시뮬레이션
일일 100만 토큰 처리 시나리오:
- GPT-4.1 전용: $8.00 × 1M/1M = $8.00/일
- Gemini 2.5 Flash 70% + GPT-4.1 30%: ($2.50 × 0.7 + $8.00 × 0.3) = $4.15/일
- DeepSeek 50% + Gemini 30% + GPT-4.1 20%: ($0.42 × 0.5 + $2.50 × 0.3 + $8.00 × 0.2) = $2.71/일
- 절감 효과: 최대 66% 비용 절감 가능
5. 이런 팀에 적합 / 비적합
✅ 이런 팀에 적합
- 높은 API 호출량: 일일 10만 건 이상 요청하는 팀
- 서비스 가용성 필수: Rate Limit으로 인한 서비스 중단이 치명적인 경우
- 비용 최적화 필요: 월간 AI API 비용이 $1,000 이상인 팀
- 다중 모델 활용:用途별 모델을 분리하여 관리하고 싶은 팀
- 해외 결제 어려움: 해외 신용카드 없이 AI API를 사용해야 하는 한국 개발자
❌ 이런 팀에는 비적합
- 소규모 개인 프로젝트: 월간 $10 미만 소비 예상
- 단일 모델 고정: 특정 모델만 사용하고 폴백이 불필요한 경우
- 완전 자체 호스팅: 모든 AI 인프라를 직접 관리하려는 팀
- 엄격한 데이터 주권: API 호출 로그도 자체 인프라에 보관해야 하는 경우
6. 가격과 ROI
HolySheep AI 요금제
| 플랜 | 월간 비용 | 선불 크레딧 | Rate Limit | 추가 기능 |
|---|---|---|---|---|
| Developer | 무료 | $5 크레딧 | 100 RPM | 기본 모델 접근 |
| Pro | $99/월 | $150 크레딧 | 500 RPM | 모든 모델, 우선 지원 |
| Enterprise | 맞춤 견적 | 협의 | 무제한 | SLA, 전담 지원, 커스텀 모델 |
ROI 계산
저는 이전에 월 $3,200의 AI API 비용을 절감한 경험이 있습니다. HolySheep의 다중 모델 라우팅을 적용한 결과:
- 기존 방식 (GPT-4 전담): 월 $4,500
- HolySheep 폴백 적용 후: 월 $1,300
- 월간 절감액: $3,200 (71% 절감)
- 회로 차단기로 인한 서비스 중단 회피: 월 12회 → 0회
7. 자주 발생하는 오류와 해결책
오류 1: HTTP 401 Unauthorized
# 증상
httpx.HTTPStatusError: 401 Client Error for url: https://api.holysheep.ai/v1/chat/completions
{"error": {"message": "Invalid API key provided", "type": "invalid_request_error"}}
원인
- 잘못된 API 키 사용
- API 키 만료
- API 키 형식 오류
해결책
import os
환경 변수로 안전하게 관리
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY environment variable not set")
또는 HolySheep 대시보드에서 새 API 키 발급
https://www.holysheep.ai/dashboard/api-keys
gateway = HolySheepGateway(api_key)오류 2: HTTP 429 Rate Limit Exceeded
# 증상
RateLimitError: Rate limit reached for gpt-4.1 in organization org-xxxxx
on requests per min. Please retry after 58 seconds
원인
- 분당 요청 수 초과 (RPM Limit)
- 분당 토큰 수 초과 (TPM Limit)
- 일일 요청 수 초과
해결책 - 지연 로직과 폴백 조합
import asyncio
from datetime import datetime, timedelta
class SmartRateLimitHandler:
"""Rate Limit을 우아하게 처리하는 핸들러"""
def __init__(self):
self.request_timestamps = []
self.token_usage = []
self.rpm_limit = 500
self.tpm_limit = 150000
async def wait_if_needed(self):
"""Rate Limit 전에 대기"""
now = datetime.now()
# 1분 내 요청 수 체크
recent_requests = [
t for t in self.request_timestamps
if (now - t).total_seconds() < 60
]
if len(recent_requests) >= self.rpm_limit:
oldest = min(recent_requests)
wait_time = 60 - (now - oldest).total_seconds()
print(f"⏳ RPM limit reached. Waiting {wait_time:.1f}s...")
await asyncio.sleep(wait_time)
self.request_timestamps.append(datetime.now())
def calculate_tokens(self, messages: List[Dict]) -> int:
"""대략적인 토큰 수估算 (실제로는 tiktoken 사용 권장)"""
total = 0
for msg in messages:
# 대략: 내용 길이 / 4 + 역할 overhead
total += len(msg.get("content", "")) // 4 + 10
return total
async def safe_request(self, func, *args, **kwargs):
"""Rate Limit 안전한 요청 실행"""
# 토큰 수 체크
if 'messages' in kwargs:
estimated_tokens = self.calculate_tokens(kwargs['messages'])
# TPM 체크
now = datetime.now()
recent_tokens = [
(t, tok) for t, tok in self.token_usage
if (now - t).total_seconds() < 60
]
current_tpm = sum(tok for _, tok in recent_tokens)
if current_tpm + estimated_tokens > self.tpm_limit:
wait_seconds = 60 - (now - min([t for t, _ in recent_tokens])).total_seconds()
print(f"⏳ TPM limit would be exceeded. Waiting {wait_seconds:.1f}s...")
await asyncio.sleep(max(wait_seconds, 1))
await self.wait_if_needed()
return await func(*args, **kwargs)오류 3: HTTP 400 Context Length Exceeded
# 증상
HTTP 400 Bad Request
{
"error": {
"message": "Maximum context window exceeded. Maximum: 128000 tokens,
Requested: 156,234 tokens",
"type": "invalid_request_error",
"code": "context_length_exceeded"
}
}
원인
- 프롬프트 + 대화 히스토리 + 출력 토큰 > 모델 컨텍스트 창
해결책 - 대화 요약 + 스마트 트렁케이션
import tiktoken
class ContextManager:
"""컨텍스트 길이 관리자"""
def __init__(self, model: str = "gpt-4.1"):
self.model = model
self.context_limits = {
"gpt-4.1": 128000,
"claude-sonnet-4-5": 200000,
"gemini-2.5-flash": 1000000,
"deepseek-v3.2": 640000
}
# 토큰 카운터 초기화
try:
self.encoder = tiktoken.encoding_for_model("gpt-4")
except:
self.encoder = tiktoken.get_encoding("cl100k_base")
def count_tokens(self, messages: List[Dict]) -> int:
"""토큰 수 계산"""
total = 0
for msg in messages:
total += 4 # role overhead
total += len(self.encoder.encode(msg.get("content", "")))
total += 2 # assistant overhead
return total
def truncate_to_fit(
self,
messages: List[Dict],
max_tokens: int = 2000,
reserve_tokens: int = 500
) -> List[Dict]:
"""적합한 크기로 트렁케이션"""
context_limit = self.context_limits.get(self.model, 128000)
available = context_limit - max_tokens - reserve_tokens
# 시스템 메시지는 항상 유지
system_messages = [m for m in messages if m.get("role") == "system"]
other_messages = [m for m in messages if m.get("role") != "system"]
result = system_messages.copy()
# 역순으로 추가 (최근 메시지 우선)
current_tokens = self.count_tokens(result)
for msg in reversed(other_messages):
msg_tokens = self.count_tokens([msg])
if current_tokens + msg_tokens <= available:
result.insert(len(system_messages), msg)
current_tokens += msg_tokens
else:
# 대화 요약으로 대체
if len(other_messages) > 2:
summary_msg = {
"role": "system",
"content": f"[{len(other_messages) - len(result) + len(system_messages)}개의 이전 메시지가省略되었습니다]"
}
result.insert(len(system_messages), summary_msg)
break
return result
def summarize_old_messages(self, messages: List[Dict]) -> List[Dict]:
"""이전 대화 요약 (단순화 버전)"""
if len(messages) <= 4:
return messages
# 시스템 메시지
system = [m for m in messages if m["role"] == "system"]
# 중간 대화 (요약 대상)
middle = messages[len(system):-2] # 마지막 2개 제외
# 최근 대화
recent = messages[-2:]
summary = {
"role":