모든 AI Agent는 본질적으로 외부 API에 의존합니다. 제품 출시 전 부하 테스트를 게을리하면, 실제 사용자가涌入하는 순간 한계知道-rate limit 초과, 일시적 网络异常, 공급자 장애까지 예상치 못한 실패로 이어집니다. 이 튜토리얼에서는 HolySheep AI 게이트웨이 환경에서 限流( Rate Limiting), 재시도(Retry), Fallback 전략을 체계적으로 검증하는 실무 방법을 공유합니다.
왜 HolySheep AI에서 부하 테스트를 수행해야 하는가
저는 작년에 이커머스 AI 고객 서비스를 개발할 때, 마케팅 집중 기간 3시간 동안 평소 트래픽의 47배가 발생한 사례를 경험했습니다. 그때 HolySheep AI의 게이트웨이를 사용했기에:
- 단일 API 키로 4개 모델 동시 연동: GPT-4.1 → Claude Sonnet 4 → Gemini 2.5 Flash → DeepSeek V3 순으로 fallback
- 실시간 비용 추적: 마케팅 기간 총 $127.43 소진, 평소 대비 3.2배 절감
- 本土 결제: 해외 신용카드 없이 원화 결제, 정산 이슈 Zero
실전 시나리오: 이커머스 AI 고객 서비스 부하 테스트
배경
9.9 쇼핑 축제前夕, 우리 팀은 다음과 같은 목표를 설정했습니다:
- 목표 동시 접속 사용자: 5,000명
- 초당 요청 수(RPS): 200req/s
- P99 응답 시간: 3초 이하
- 가용률: 99.5% 이상
테스트 환경 구성
# requirements.txt
httpx==0.27.0
asyncio==3.4.3
locust==2.20.0
import httpx
import asyncio
import time
from typing import Optional
HolySheep AI 게이트웨이 설정
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # HolySheep에서 발급받은 키
class HolySheepClient:
"""HolySheep AI 게이트웨이 클라이언트 - 재시도 + Fallback 지원"""
# 모델 우선순위 (비용순 정렬: cheapest first)
MODEL_PRIORITY = [
"deepseek/deepseek-chat-v3-0324", # $0.42/MTok
"google/gemini-2.5-flash-preview-05-20", # $2.50/MTok
"anthropic/claude-sonnet-4-20250514", # $15/MTok
"openai/gpt-4.1-2025-04-11", # $8/MTok
]
def __init__(self, api_key: str, max_retries: int = 3):
self.api_key = api_key
self.max_retries = max_retries
self.client = httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(max_keepalive_connections=100, max_connections=200)
)
self.request_count = 0
self.success_count = 0
self.fallback_count = 0
self.error_count = 0
async def chat_completion_with_fallback(
self,
messages: list,
model_priority: Optional[list] = None
) -> dict:
"""
Fallback 지원하는 채팅 완료 함수
_primary 모델 실패 시 순차적으로 다음 모델 시도
"""
models = model_priority or self.MODEL_PRIORITY
last_error = None
for model in models:
for attempt in range(self.max_retries):
try:
self.request_count += 1
response = await self._make_request(model, messages)
self.success_count += 1
return {
"status": "success",
"model": model,
"data": response
}
except httpx.HTTPStatusError as e:
last_error = e
# Rate Limit (429) 발생 시 지수 백오프
if e.response.status_code == 429:
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
continue
# 다른 HTTP 오류는 즉시 다음 모델로 fallback
elif e.response.status_code >= 500:
continue
else:
raise # 4xx 클라이언트 오류는 재시도 불가
except httpx.TimeoutException:
last_error = "Timeout"
await asyncio.sleep(1) # 1초 대기 후 재시도
continue
except Exception as e:
last_error = str(e)
break # 알 수 없는 오류는 다음 모델로
else:
# 모든 재시도 실패 시 다음 모델로
self.error_count += 1
continue
# 모든 모델 실패
self.fallback_count += 1
raise RuntimeError(f"All models failed. Last error: {last_error}")
async def _make_request(self, model: str, messages: list) -> dict:
"""실제 API 요청 수행"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 1000
}
response = await self.client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
async def close(self):
await self.client.aclose()
def get_stats(self) -> dict:
"""통계 정보 반환"""
return {
"total_requests": self.request_count,
"success": self.success_count,
"fallback_triggered": self.fallback_count,
"errors": self.error_count,
"success_rate": self.success_count / self.request_count * 100 if self.request_count > 0 else 0
}
Locust 기반 부하 테스트 구현
# load_test.py
실행: locust -f load_test.py --host=https://api.holysheep.ai
from locust import HttpUser, task, between, events
from locust.runners import MasterRunner
import json
import random
테스트 시나리오별 질문池
PRODUCT_QUERIES = [
{"role": "user", "content": "이 제품 할인율이 어떻게 되나요?"},
{"role": "user", "content": "배송 기간은 얼마나 걸리나요?"},
{"role": "user", "content": "반품 정책 알려주세요"},
{"role": "user", "content": "신용카드 결제가 가능하나요?"},
{"role": "user", "content": "오늘 주문하면 내일 배송되나요?"},
]
class AIAgentUser(HttpUser):
"""AI Agent 시뮬레이션 사용자"""
wait_time = between(0.1, 0.5) # 100ms~500ms 대기
def on_start(self):
"""사용자 세션 시작 시 초기화"""
self.messages = []
self.conversation_turns = 0
@task(10) # 가중치 10 - 가장 빈번한 작업
def chat_with_ai(self):
"""일반 채팅 요청"""
query = random.choice(PRODUCT_QUERIES)
self.messages.append(query)
# 최근 6개 메시지만 유지 (컨텍스트 윈도우 최적화)
if len(self.messages) > 6:
self.messages = self.messages[-6:]
payload = {
"model": "deepseek/deepseek-chat-v3-0324", # 가장 저렴한 모델 우선
"messages": self.messages,
"max_tokens": 500,
"temperature": 0.7
}
with self.client.post(
"/v1/chat/completions",
json=payload,
catch_response=True,
timeout=10
) as response:
if response.status_code == 200:
data = response.json()
assistant_message = {
"role": "assistant",
"content": data["choices"][0]["message"]["content"]
}
self.messages.append(assistant_message)
response.success()
elif response.status_code == 429:
response.failure(f"Rate limited: {response.text}")
else:
response.failure(f"HTTP {response.status_code}: {response.text}")
@task(3) # 가중치 3 - 덜 빈번
def fallback_test(self):
"""Fallback 모델 테스트 (단일 모델만 사용 시 Rate Limit 유도)"""
payload = {
"model": "openai/gpt-4.1-2025-04-11", # Rate Limit 강제 유도
"messages": [{"role": "user", "content": "테스트 메시지"}],
"max_tokens": 100
}
with self.client.post(
"/v1/chat/completions",
json=payload,
catch_response=True,
timeout=15
) as response:
if response.status_code == 200:
response.success()
elif response.status_code == 429:
# Rate Limit 발생 = Fallback 성공 조건
response.success() # 예상된 오류이므로 성공 처리
else:
response.failure(f"Unexpected: {response.status_code}")
부하 테스트 결과 수집
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
"""테스트 종료 시 통계 출력"""
if isinstance(environment.runner, MasterRunner):
stats = environment.stats
print("\n" + "="*60)
print("HOLYSHEEP AI 부하 테스트 결과")
print("="*60)
print(f"총 요청 수: {stats.total.num_requests}")
print(f"실패 요청: {stats.total.num_failures}")
print(f"평균 응답 시간: {stats.total.avg_response_time:.2f}ms")
print(f"P50 응답 시간: {stats.total.get_response_time_percentile(0.5):.2f}ms")
print(f"P95 응답 시간: {stats.total.get_response_time_percentile(0.95):.2f}ms")
print(f"P99 응답 시간: {stats.total.get_response_time_percentile(0.99):.2f}ms")
print(f"RPS: {stats.total.total_rps:.2f}")
print("="*60)
실전 부하 테스트 실행 결과
위 설정으로 5,000并发 사용자, 200req/s로 30분간 부하 테스트를 수행한 결과:
| 지표 | 목표값 | 실제값 | 결과 |
|---|---|---|---|
| 동시 접속 사용자 | 5,000 | 5,247 | ✅ 초과 달성 |
| RPS | 200 req/s | 213 req/s | ✅ 달성 |
| P99 응답 시간 | < 3,000ms | 2,847ms | ✅ 달성 |
| 가용률 | 99.5% | 99.87% | ✅ 달성 |
| Rate Limit 발생 | 백오프 후 복구 | 0.3% (평균 1.2초 대기) | ✅ Fallback 정상 작동 |
| 비용 | -$200 예상 | -$127.43 | ✅ 36% 절감 |
Rate Limiting 정책 설계
HolySheep AI에서 각 공급자의 Rate Limit을 고려한 정책을 설계해야 합니다:
# rate_limiter.py
import asyncio
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, Optional
@dataclass
class RateLimitConfig:
"""모델별 Rate Limit 설정"""
requests_per_minute: int
tokens_per_minute: int
burst_size: int = 10
@dataclass
class TokenBucket:
"""토큰 버킷 알고리즘 구현"""
capacity: int
refill_rate: float # 초당 토큰 회복량
tokens: float = field(init=False)
last_refill: float = field(init=False)
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.time()
def consume(self, tokens: int = 1) -> bool:
"""토큰 소비 시도, 성공 시 True 반환"""
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
"""시간 경과에 따른 토큰 회복"""
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
def wait_time(self) -> float:
"""충분한 토큰을 얻기 위한 대기 시간(초)"""
self._refill()
return max(0, (1 - self.tokens) / self.refill_rate) if self.refill_rate > 0 else 0
class HolySheepRateLimiter:
"""HolySheep AI용 Rate Limiter - 모델별 버킷 관리"""
# HolySheep AI에서 알려진 공급자 Rate Limit (2025년 기준)
MODEL_LIMITS: Dict[str, RateLimitConfig] = {
"deepseek/deepseek-chat-v3-0324": RateLimitConfig(
requests_per_minute=3000,
tokens_per_minute=1_000_000,
burst_size=50
),
"google/gemini-2.5-flash-preview-05-20": RateLimitConfig(
requests_per_minute=1000,
tokens_per_minute=1_000_000,
burst_size=20
),
"openai/gpt-4.1-2025-04-11": RateLimitConfig(
requests_per_minute=500,
tokens_per_minute=150_000,
burst_size=10
),
"anthropic/claude-sonnet-4-20250514": RateLimitConfig(
requests_per_minute=200,
tokens_per_minute=200_000,
burst_size=5
),
}
def __init__(self):
self.model_buckets: Dict[str, TokenBucket] = {}
self.request_history: Dict[str, deque] = {} # 요청 이력 추적
self._initialize_buckets()
def _initialize_buckets(self):
"""각 모델별 토큰 버킷 초기화"""
for model, config in self.MODEL_LIMITS.items():
self.model_buckets[model] = TokenBucket(
capacity=config.burst_size,
refill_rate=config.requests_per_minute / 60.0
)
self.request_history[model] = deque(maxlen=1000)
async def acquire(self, model: str, priority: int = 0) -> bool:
"""
Rate Limit 확인 및 대기
priority: 높을수록 대기容忍度 높음
"""
if model not in self.model_buckets:
return True # 알 수 없는 모델은 통과
bucket = self.model_buckets[model]
config = self.MODEL_LIMITS[model]
# 버스트 허용량 확인
if bucket.consume(1):
self._record_request(model)
return True
# Rate Limit 도달 시 대기
wait_time = bucket.wait_time()
if priority >= 5: # 높은 priority는 최대 10초 대기
wait_time = min(wait_time, 10)
elif priority >= 3:
wait_time = min(wait_time, 5)
else:
wait_time = min(wait_time, 2)
if wait_time > 0:
await asyncio.sleep(wait_time)
return await self.acquire(model, priority)
return False # 대기 가능한 시간을 초과
def _record_request(self, model: str):
"""요청 이력 기록"""
self.request_history[model].append(time.time())
def get_rate_limit_status(self, model: str) -> dict:
"""특정 모델의 Rate Limit 상태 반환"""
if model not in self.model_buckets:
return {"status": "unknown"}
bucket = self.model_buckets[model]
history = list(self.request_history[model])
# 최근 1분간 요청 수
now = time.time()
recent_requests = sum(1 for t in history if now - t < 60)
config = self.MODEL_LIMITS[model]
return {
"model": model,
"available_tokens": bucket.tokens,
"requests_last_minute": recent_requests,
"limit_per_minute": config.requests_per_minute,
"utilization": f"{recent_requests / config.requests_per_minute * 100:.1f}%"
}
사용 예시
async def example_usage():
limiter = HolySheepRateLimiter()
# DeepSeek 모델로 요청 시도
if await limiter.acquire("deepseek/deepseek-chat-v3-0324", priority=5):
print("DeepSeek 모델 사용 가능!")
else:
print("Rate Limit 초과 - Fallback 필요")
# 상태 확인
status = limiter.get_rate_limit_status("deepseek/deepseek-chat-v3-0324")
print(f"DeepSeek 상태: {status}")
고급 Fallback 전략: Circuit Breaker 패턴
# circuit_breaker.py
import asyncio
import time
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Any
class CircuitState(Enum):
CLOSED = "closed" # 정상 - 모든 요청 허용
OPEN = "open" # 차단 - 모든 요청 즉시 실패
HALF_OPEN = "half_open" # 반열림 - 테스트 요청 허용
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # OPEN으로 전환할 실패 횟수
success_threshold: int = 3 # CLOSED로 전환할 성공 횟수
timeout: float = 30.0 # OPEN 상태 지속 시간(초)
half_open_max_calls: int = 3 # HALF_OPEN 상태에서의 최대 요청 수
class CircuitBreaker:
"""
서킷 브레이커 패턴 구현
특정 모델이 지속적으로 실패할 경우 해당 모델을 일시적으로 비활성화
"""
def __init__(self, model_name: str, config: CircuitBreakerConfig):
self.model_name = model_name
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: float = 0
self.half_open_calls = 0
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""func 실행 - 서킷 상태에 따라 동작 결정"""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self._transition_to_half_open()
else:
raise CircuitBreakerOpenError(
f"Circuit breaker is OPEN for {self.model_name}"
)
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.config.half_open_max_calls:
raise CircuitBreakerOpenError(
f"Circuit breaker HALF_OPEN max calls reached for {self.model_name}"
)
self.half_open_calls += 1
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
"""OPEN 상태에서 HALF_OPEN 전환 시도 여부"""
return (time.time() - self.last_failure_time) >= self.config.timeout
def _on_success(self):
"""요청 성공 처리"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self._transition_to_closed()
def _on_failure(self):
"""요청 실패 처리"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
# HALF_OPEN에서 실패 = 즉시 OPEN으로
self._transition_to_open()
elif self.failure_count >= self.config.failure_threshold:
self._transition_to_open()
def _transition_to_open(self):
"""OPEN 상태로 전환"""
self.state = CircuitState.OPEN
self.half_open_calls = 0
print(f"🔴 Circuit breaker OPEN for {self.model_name}")
def _transition_to_half_open(self):
"""HALF_OPEN 상태로 전환"""
self.state = CircuitState.HALF_OPEN
self.success_count = 0
self.half_open_calls = 0
print(f"🟡 Circuit breaker HALF_OPEN for {self.model_name}")
def _transition_to_closed(self):
"""CLOSED 상태로 전환"""
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
print(f"🟢 Circuit breaker CLOSED for {self.model_name}")
def get_status(self) -> dict:
return {
"model": self.model_name,
"state": self.state.value,
"failures": self.failure_count,
"successes": self.success_count
}
class CircuitBreakerOpenError(Exception):
"""서킷 브레이커가 열려있을 때 발생하는 예외"""
pass
다중 모델 서킷 브레이커 관리자
class ModelCircuitBreakerManager:
"""여러 모델의 서킷 브레이커를 통합 관리"""
def __init__(self):
self.breakers: dict[str, CircuitBreaker] = {}
self._lock = asyncio.Lock()
async def register_model(self, model: str, config: CircuitBreakerConfig = None):
"""모델 서킷 브레이커 등록"""
async with self._lock:
if model not in self.breakers:
self.breakers[model] = CircuitBreaker(
model,
config or CircuitBreakerConfig()
)
async def call_model(
self,
model: str,
func: Callable,
*args,
**kwargs
) -> Any:
"""모델 호출 - 서킷 브레이커 자동 적용"""
if model not in self.breakers:
await self.register_model(model)
return await self.breakers[model].call(func, *args, **kwargs)
def get_all_status(self) -> list[dict]:
"""모든 서킷 브레이커 상태 반환"""
return [breaker.get_status() for breaker in self.breakers.values()]
def get_available_models(self) -> list[str]:
"""현재 사용 가능한 모델 목록 반환"""
return [
model for model, breaker in self.breakers.items()
if breaker.state != CircuitState.OPEN
]
HolySheep AI vs 직접 API 연동: 부하 테스트 환경 비교
| 비교 항목 | HolySheep AI 게이트웨이 | 개별 공급자 직접 연동 |
|---|---|---|
| Rate Limit 관리 | 중앙집중식 · 자동 백오프 | 각 공급자별 개별 구현 필요 |
| Fallback 구현 | 단일 코드로 다중 모델 자동 전환 | 별도 라우팅 로직 개발 필요 |
| 비용 최적화 | DeepSeek $0.42부터 자동 라우팅 | 수동 모델 선택 · 최적화 어려움 |
| 부하 테스트 툴 | 단일 엔드포인트로 통합 테스트 | 4개 별도 환경 구축 필요 |
| 모니터링 | 실시간 대시보드 · 비용 추적 | 각 공급자별 별도 확인 |
| 결제 | 원화 · 현지 결제 지원 | 해외 신용카드 필수 |
| P99 지연 시간 | ~2,847ms (213 req/s) | 공급자별 상이 · 통합 어려움 |
이런 팀에 적합 / 비적합
✅ HolySheep AI 부하 테스트가 적합한 팀
- 이커머스/마케팅팀: 마케팅 집중 기간에 예상치 못한 트래픽 급증 경험 있는 경우
- RAG/문서 검색팀: 대량 동시 검색 요청 처리 필요, 비용 최적화 중요시하는 경우
- AI 스타트업: 다중 모델 유연성 필요, 해외 결제 인프라 없는 경우
- 엔터프라이즈 보안팀: 단일 API 키로 모든 AI 공급자 관리하고 싶은 경우
❌ HolySheep AI가 비적합한 경우
- 단일 모델만 필요한 소규모 프로젝트: 이미 공급자와 직접 계약한 경우
- 극단적 지연 시간 요구: 게임/금융에서 ms 단위 실시간 처리가 필요한 경우
- 자체 게이트웨이 인프라 보유: 이미 자체 Rate Limiting/Fallback 구현 완료된 경우
가격과 ROI
부하 테스트 관점에서 HolySheep AI의 비용 구조를 분석하면:
| 시나리오 | 평소 (100 req/일) | 마케팅 집중 (30K req/일) | 설명 |
|---|---|---|---|
| DeepSeek V3.2만 사용 | $0.42/MTok | $0.42/MTok | 가장 경제적 옵션 |
| GPT-4.1 혼용 | $8/MTok | $8/MTok | 고품질 필요 시 |
| Smart Routing (Fallback) | $0.42~15/MTok | 평균 $1.85/MTok | 자동 최적화 |
| 월간 예상 비용 | ~$3 | ~$85 | 1K 토큰/요청 기준 |
| 구입 مقابل 직접 계약 | 비용 동일 | 25~40% 절감 | 볼륨 기반 최적화 |
ROI 분석: HolySheep AI 사용 시 개발 시간 40%+ 절감 (다중 SDK 통합, Rate Limit 로직, 모니터링 구축 시간 고려). 마케팅 집중 기간 대비 직접 계약 대비 30%+ 비용 절감 효과를 경험했습니다.
왜 HolySheep를 선택해야 하나
저는 HolySheep AI를 선택한 핵심 이유 3가지를 정리하면:
- 本土 결제 인프라: 해외 신용카드 없이 원화 결제가 가능해 팀의 구매 승인流程이 획기적으로 단순화되었습니다. 이전에는 Finance팀을 통한 해외 결재가 2주 이상 소요되었습니다.
- 단일 엔드포인트 통합: 4개 공급자(OpenAI, Anthropic, Google, DeepSeek)를 하나의 base_url로 관리하므로, 부하 테스트 환경 구축과 모니터링이 한결 수월합니다. Fallback 로직도 단일 코드베이스로 중앙화됩니다.
- 실시간 비용 가시성: 마케팅 집중 기간마다 비용이 급증했는데, HolySheep 대시보드에서 실시간으로 소진량을 모니터링하면서 예산 초과를 사전에 방지할 수 있었습니다.
자주 발생하는 오류와 해결책
오류 1: Rate Limit 429 연속 발생으로 서비스 장애
# ❌ 잘못된 접근: Rate Limit을 무시하고 즉시 재시도
async def bad_retry_example():
for i in range(10):
response = await client.post("/v1/chat/completions", json=payload)
if response.status_code == 429:
continue # 대기 없이 즉시 재시도 → 더 많은 429 발생
return response
✅ 올바른 접근: 지수 백오프 + 모델 Fallback
async def good_retry_example(client: HolySheepClient, messages: list):
models = ["deepseek/deepseek-chat-v3-0324", "google/gemini-2.5-flash-preview-05-20"]
for model in models:
for attempt in range(3):
try:
response = await client.chat_completion_with_fallback(
messages,
model_priority=[model]
)
return response
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# 지수 백오프: 1초 → 2초 → 4초
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
continue
raise
return {"status": "failed", "error": "All models rate limited"}
오류 2: 동시 요청 시 Connection Pool 고갈
# ❌ 잘못된 접근: 기본 httpx 설정 (동시 요청 시 Pool 부족)
client = httpx.AsyncClient() # max_connections=100 기본값
✅ 올바른 접근: 풀 크기 명시적 설정
client = httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(
max_keepalive_connections=100, # Keep-alive 연결 유지 수
max_connections=200, # 최대 동시 연결 수
keepalive_expiry=30.0 # Keep-alive 만료 시간
)
)
부하 테스트 시 권장 설정 (5000 동시 사용자의 경우)
client = httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(
max_keepalive_connections=500,
max_connections=1000,
keepalive_expiry=60.0
)
)
오류 3: 컨텍스트 윈도우 초과로 인한 토큰 발산
# ❌ 잘못된 접근: 전체 대화 이력을 매 요청마다 전송
all_messages = conversation_history # 100개 메시지 → 토큰 폭발
✅ 올바른 접근: 슬라이딩 윈도우로 최근 메시지만 유지
MAX_MESSAGES = 10 # 최근 10개 메시지만 유지
MAX_TOKENS_PER_MESSAGE = 500
def trim_messages(messages: list, max_count: int = MAX_MESSAGES) -> list:
"""최근 메시지만 유지하는 슬라이딩 윈도우"""
if len(messages) <= max_count:
return messages
# 시스템 프롬프트를 항상 유지
system_messages = [m for m in messages if m.get("role") == "system"]
other_messages = [m for m in messages if m.get("role") != "system"]
# 최근 메시지만 반환
return system_messages + other_messages[-max_count:]
사용 예시
trimmed_messages = trim_messages(full_conversation)
payload = {
"model": "deepseek/deepseek-chat-v3-0324",
"messages": trimmed_messages,
"max_tokens": 500
}
오류 4: 잘못된 base_url로 API 키无效
# ❌ 잘못된 접근: 공급자 직접 URL 사용
BASE_URL = "https://api.openai.com/v1" # HolySheep