핵심 결론부터 확인하세요
Mock Testing은 선택이 아닌 필수입니다. AI API 호출 비용은 GPT-4.1 기준 $8/1M 토큰, Claude Sonnet 4.5 기준 $15/1M 토큰에 달합니다. 개발 단계에서 실제 API를 호출하면 하루에 수십 달러가 순식간에 사라질 수 있습니다. 이 튜토리얼에서는 HolySheep AI를 포함한 모든 AI API에서 안전하게 Mock Testing을 구현하는 방법을 단계별로 설명드리겠습니다.
저는 3년 넘게 AI API 통합 프로젝트를 진행하면서 수천 달러의 비용을 절감한 경험이 있습니다. 이 글에서 소개하는 방법들을 적용하면 개발 환경에서 실제 API 비용을 100% 절감하면서도 프로덕션 환경과 동일한 코드 구조를 유지할 수 있습니다.
왜 AI API Mock Testing이 중요한가?
- 비용 최적화: 개발/QA 환경에서 실제 API 호출 비용为零 (0)
- скорость (속도): 네트워크 지연 없이 밀리초 단위 응답 반환
- 격리된 테스트: 외부 의존성 없이 단위 테스트와 통합 테스트 가능
- 일관된 결과: 결정적(deterministic) 응답으로 테스트 신뢰성 확보
- 시나리오 커버리지: 에러 케이스, 타임아웃, 레이트 리밋 등 다양한 상황 시뮬레이션
AI API 게이트웨이 비교표
| 서비스 | 결제 방식 | GPT-4.1 | Claude Sonnet 4.5 | Gemini 2.5 Flash | DeepSeek V3.2 | 적합한 팀 |
|---|---|---|---|---|---|---|
| HolySheep AI | 로컬 결제 (해외 카드 불필요) | $8/MTok | $15/MTok | $2.50/MTok | $0.42/MTok | 글로벌 서비스 필요 팀, 스타트업 |
| OpenAI 공식 | 해외 신용카드 필수 | $8/MTok | - | - | - | OpenAI 전용 프로젝트 |
| Anthropic 공식 | 해외 신용카드 필수 | - | $15/MTok | - | - | Claude 전용 프로젝트 |
| Google Vertex AI | 해외 신용카드 필수 | - | - | $2.50/MTok | - | Enterprise GCP 사용자 |
HolySheep AI는 단일 API 키로 모든 주요 모델을 통합 지원하며, 해외 신용카드 없이 로컬 결제가 가능합니다. 개발 단계에서 Mock Testing 후 동일 코드베이스로 실제 API切り替え가无缝합니다.
Python 기반 Mock Testing 구현
1. 기본 Mock 클라이언트 클래스
import json
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
@dataclass
class MockMessage:
role: str
content: str
@dataclass
class MockChoice:
index: int
message: MockMessage
finish_reason: str
@dataclass
class MockUsage:
prompt_tokens: int
completion_tokens: int
total_tokens: int
@dataclass
class MockResponse:
id: str
object: str = "chat.completion"
created: int = field(default_factory=lambda: int(time.time()))
model: str = "mock-model"
choices: List[MockChoice] = field(default_factory=list)
usage: Optional[MockUsage] = None
class AIAPIMockClient:
"""AI API Mock Testing을 위한 범용 클라이언트"""
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
self.response_delay_ms: int = 0
self.custom_responses: Dict[str, str] = {}
self.error_rate: float = 0.0
self._call_count: int = 0
def set_response_delay(self, delay_ms: int) -> "AIAPIMockClient":
"""응답 지연 시간 설정 (밀리초 단위)"""
self.response_delay_ms = delay_ms
return self
def set_custom_response(self, prompt_pattern: str, response: str) -> "AIAPIMockClient":
"""특정 프롬프트 패턴에 대한 커스텀 응답 설정"""
self.custom_responses[prompt_pattern] = response
return self
def set_error_rate(self, rate: float) -> "AIAPIMockClient":
"""에러 발생 확률 설정 (0.0 ~ 1.0)"""
if not 0.0 <= rate <= 1.0:
raise ValueError("error_rate must be between 0.0 and 1.0")
self.error_rate = rate
return self
def chat_completions(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000,
**kwargs
) -> Dict[str, Any]:
"""
Mock Chat Completions API
실제 HolySheep AI API 응답 형식과 100% 호환
"""
self._call_count += 1
# 응답 지연 시뮬레이션
if self.response_delay_ms > 0:
time.sleep(self.response_delay_ms / 1000.0)
# 마지막 사용자 메시지 추출
last_message = messages[-1]["content"] if messages else ""
# 커스텀 응답 확인
response_content = self._get_response_content(last_message, model)
# Mock 응답 생성
response = {
"id": f"mock-{self._call_count}-{int(time.time())}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": response_content
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": len(str(messages)) // 4,
"completion_tokens": len(response_content) // 4,
"total_tokens": (len(str(messages)) + len(response_content)) // 4
}
}
return response
def _get_response_content(self, prompt: str, model: str) -> str:
"""프롬프트에 맞는 응답 콘텐츠 반환"""
for pattern, response in self.custom_responses.items():
if pattern.lower() in prompt.lower():
return response
# 기본 Mock 응답
return f"[Mock Response for: {prompt[:50]}... | Model: {model}]"
사용 예시
if __name__ == "__main__":
client = AIAPIMockClient()
# 150ms 지연 설정 (실제 API 평균 지연 시간 시뮬레이션)
client.set_response_delay(150)
# 커스텀 응답 설정
client.set_custom_response(" translate ", "번역 결과: Hello World")
client.set_custom_response(" summarize ", "요약: 이 문장의 핵심 내용은...")
# API 호출
response = client.chat_completions(
model="gpt-4.1",
messages=[{"role": "user", "content": "Please translate to Korean: Hello World"}]
)
print(f"Response ID: {response['id']}")
print(f"Content: {response['choices'][0]['message']['content']}")
print(f"Total Tokens: {response['usage']['total_tokens']}")
2. 환경별 Mock/실제 API 자동 전환
import os
from typing import Optional
from abc import ABC, abstractmethod
class BaseAIClient(ABC):
"""AI API 클라이언트 추상 기본 클래스"""
@abstractmethod
def chat_completions(self, model: str, messages: list, **kwargs):
pass
class MockAIClient(BaseAIClient):
"""Mock 환경용 클라이언트 - HolySheep AI 포맷 호환"""
def __init__(self):
from collections import defaultdict
self._responses = defaultdict(lambda: {
"id": "mock-response",
"object": "chat.completion",
"created": 1234567890,
"model": "mock",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "Mock response"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
})
def add_response(self, model: str, content: str) -> None:
"""모델별 고정 응답 추가"""
import time
self._responses[model] = {
"id": f"mock-{model}-{int(time.time() * 1000)}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": content},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 20, "completion_tokens": len(content.split()), "total_tokens": 20 + len(content.split())}
}
def chat_completions(self, model: str, messages: list, **kwargs):
return self._responses[model]
class HolySheepAIClient(BaseAIClient):
"""HolySheep AI 실제 API 클라이언트"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self._session = None
def _get_session(self):
if self._session is None:
import requests
self._session = requests.Session()
self._session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
return self._session
def chat_completions(self, model: str, messages: list, **kwargs):
import requests
import time
start_time = time.time()
try:
response = self._get_session().post(
f"{self.BASE_URL}/chat/completions",
json={
"model": model,
"messages": messages,
**{k: v for k, v in kwargs.items() if v is not None}
},
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
latency_ms = int((time.time() - start_time) * 1000)
raise RuntimeError(f"HolySheep AI API Error (Latency: {latency_ms}ms): {str(e)}")
class AIAgentFactory:
"""환경에 따라 Mock/실제 API 자동 선택"""
@staticmethod
def create_client(env: Optional[str] = None) -> BaseAIClient:
"""
환경에 맞는 AI 클라이언트 생성
Args:
env: 'development', 'testing', 'production' 또는 None (자동 감지)
Returns:
BaseAIClient 인스턴스
"""
if env is None:
env = os.getenv("AI_API_ENV", "development")
if env in ("development", "testing", "local", "ci"):
print(f"[Mock Mode] {env} 환경에서 Mock API 사용")
return MockAIClient()
elif env == "production":
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다.")
print(f"[Production Mode] HolySheep AI API 사용")
return HolySheepAIClient(api_key)
else:
raise ValueError(f"알 수 없는 환경: {env}")
테스트 코드
if __name__ == "__main__":
# Mock 모드 테스트
mock_client = AIAgentFactory.create_client("testing")
mock_client.add_response("gpt-4.1", "이것은 Mock 응답입니다.")
result = mock_client.chat_completions(
model="gpt-4.1",
messages=[{"role": "user", "content": "테스트 메시지"}]
)
print(f"Mock Result: {result['choices'][0]['message']['content']}")
# 환경 변수 설정 시뮬레이션
os.environ["AI_API_ENV"] = "production"
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
# 실제 API 전환 테스트 (실제 호출은 주석 처리)
# real_client = AIAgentFactory.create_client("production")
# result = real_client.chat_completions(model="gpt-4.1", messages=[...])
Node.js/TypeScript Mock Testing 구현
/**
* AI API Mock Testing - TypeScript 구현
* HolySheep AI API 응답 포맷 100% 호환
*/
interface AIMessage {
role: 'system' | 'user' | 'assistant';
content: string;
}
interface ChatCompletionRequest {
model: string;
messages: AIMessage[];
temperature?: number;
max_tokens?: number;
top_p?: number;
frequency_penalty?: number;
presence_penalty?: number;
stop?: string | string[];
}
interface ChatCompletionChoice {
index: number;
message: AIMessage;
finish_reason: string;
}
interface Usage {
prompt_tokens: number;
completion_tokens: number;
total_tokens: number;
}
interface ChatCompletionResponse {
id: string;
object: 'chat.completion';
created: number;
model: string;
choices: ChatCompletionChoice[];
usage: Usage;
}
type AIAuthFunction = (
req: ChatCompletionRequest
) => ChatCompletionResponse | Promise;
class MockAIClient {
private responses: Map = new Map();
private delayMs: number = 0;
private callCount: number = 0;
private authFunctions: AIAuthFunction[] = [];
setDelay(delayMs: number): MockAIClient {
this.delayMs = delayMs;
return this;
}
setResponse(model: string, response: string): MockAIClient {
this.responses.set(model, response);
return this;
}
addAuthFunction(fn: AIAuthFunction): MockAIClient {
this.authFunctions.push(fn);
return this;
}
async chatCompletions(
request: ChatCompletionRequest
): Promise {
this.callCount++;
// 지연 시간 시뮬레이션
if (this.delayMs > 0) {
await new Promise(resolve => setTimeout(resolve, this.delayMs));
}
// 커스텀 인증 함수 우선 실행
for (const authFn of this.authFunctions) {
const result = await authFn(request);
if (result) {
return {
...result,
id: mock-${this.callCount}-${Date.now()},
created: Math.floor(Date.now() / 1000)
};
}
}
// 기본 Mock 응답 생성
const lastMessage = request.messages[request.messages.length - 1]?.content || '';
const mockContent = this.responses.get(request.model) ||
[Mock] Request: ${lastMessage.substring(0, 50)}...;
return {
id: mock-${this.callCount}-${Date.now()},
object: 'chat.completion',
created: Math.floor(Date.now() / 1000),
model: request.model,
choices: [{
index: 0,
message: {
role: 'assistant',
content: mockContent
},
finish_reason: 'stop'
}],
usage: {
prompt_tokens: Math.floor(lastMessage.length / 4),
completion_tokens: Math.floor(mockContent.length / 4),
total_tokens: Math.floor((lastMessage.length + mockContent.length) / 4)
}
};
}
getCallCount(): number {
return this.callCount;
}
reset(): void {
this.callCount = 0;
this.responses.clear();
this.authFunctions = [];
}
}
// HolySheep AI 실제 API 클라이언트
class HolySheepAIClient {
private baseUrl = 'https://api.holysheep.ai/v1';
private apiKey: string;
constructor(apiKey: string) {
this.apiKey = apiKey;
}
async chatCompletions(
request: ChatCompletionRequest,
timeoutMs: number = 30000
): Promise {
const startTime = performance.now();
const response = await fetch(${this.baseUrl}/chat/completions, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': Bearer ${this.apiKey}
},
body: JSON.stringify(request),
signal: AbortSignal.timeout(timeoutMs)
});
const latencyMs = Math.round(performance.now() - startTime);
if (!response.ok) {
const error = await response.text();
throw new Error(
HolySheep AI Error [${response.status}] [Latency: ${latencyMs}ms]: ${error}
);
}
console.log([HolySheep AI] Latency: ${latencyMs}ms | Model: ${request.model});
return response.json();
}
}
// 환경별 클라이언트 팩토리
function createAIClient(env?: string): MockAIClient | HolySheepAIClient {
const environment = env || process.env.AI_API_ENV || 'development';
switch (environment) {
case 'development':
case 'testing':
case 'local':
case 'ci':
console.log([Mock Mode] ${environment} 환경);
return new MockAIClient();
case 'production':
const apiKey = process.env.HOLYSHEEP_API_KEY;
if (!apiKey) {
throw new Error('HOLYSHEEP_API_KEY 환경 변수가 필요합니다.');
}
console.log('[Production Mode] HolySheep AI API 사용');
return new HolySheepAIClient(apiKey);
default:
throw new Error(알 수 없는 환경: ${environment});
}
}
// 사용 예시
async function main() {
const client = createAIClient('testing');
if (client instanceof MockAIClient) {
client
.setDelay(120) // 120ms 지연 시뮬레이션
.setResponse('gpt-4.1', 'Mock: 안녕하세요! 무엇을 도와드릴까요?')
.setResponse('claude-sonnet-4.5', 'Mock: Hello! How can I assist you?');
}
const response = await client.chatCompletions({
model: 'gpt-4.1',
messages: [
{ role: 'user', content: '인사해 주세요' }
],
temperature: 0.7,
max_tokens: 500
});
console.log('Response ID:', response.id);
console.log('Content:', response.choices[0].message.content);
console.log('Tokens:', response.usage.total_tokens);
}
main().catch(console.error);
export { MockAIClient, HolySheepAIClient, createAIClient };
export type { ChatCompletionRequest, ChatCompletionResponse };
에지 케이스 및 시나리오별 Mock 전략
"""
고급 Mock Testing 시나리오
에러 처리, 레이트 리밋, 타임아웃 등 다양한 상황 시뮬레이션
"""
import random
import time
from typing import Callable, Optional
from dataclasses import dataclass
from enum import Enum
class MockErrorType(Enum):
RATE_LIMIT = "rate_limit"
TIMEOUT = "timeout"
SERVER_ERROR = "server_error"
AUTH_ERROR = "auth_error"
INVALID_REQUEST = "invalid_request"
NETWORK_ERROR = "network_error"
@dataclass
class MockError(Exception):
error_type: MockErrorType
message: str
status_code: int = 429
def to_openai_format(self) -> dict:
return {
"error": {
"message": self.message,
"type": self.error_type.value,
"code": self.error_type.value,
"status": self.status_code
}
}
class AdvancedMockClient:
"""고급 Mock Testing 클라이언트 - 다양한 에러 시나리오 지원"""
def __init__(self):
self.call_history: list = []
self.error_config: dict = {}
self.rate_limit_remaining: int = 100
self.rate_limit_reset: float = 0
self.latency_ms: int = 0
def configure_error_scenario(
self,
error_type: MockErrorType,
probability: float,
message: Optional[str] = None
) -> "AdvancedMockClient":
"""에러 발생 시나리오 설정"""
if not 0.0 <= probability <= 1.0:
raise ValueError("probability must be between 0.0 and 1.0")
self.error_config[error_type] = {
"probability": probability,
"message": message or f"Mock {error_type.value} occurred"
}
return self
def set_rate_limit(self, remaining: int, reset_timestamp: float) -> "AdvancedMockClient":
"""레이트 리밋 설정"""
self.rate_limit_remaining = remaining
self.rate_limit_reset = reset_timestamp
return self
def set_latency(self, ms: int) -> "AdvancedMockClient":
"""평균 지연 시간 설정 (실제 API 지연 시뮬레이션)"""
self.latency_ms = ms
return self
def add_to_history(self, request: dict, response: dict) -> None:
"""호출 이력 기록"""
self.call_history.append({
"timestamp": time.time(),
"request": request,
"response": response
})
def chat_completions(self, model: str, messages: list, **kwargs):
"""Mock Chat Completions with advanced scenarios"""
# 지연 시간 적용
if self.latency_ms > 0:
time.sleep(self.latency_ms / 1000.0)
# 요청 기록
request = {"model": model, "messages": messages, **kwargs}
# 레이트 리밋 체크
if self.rate_limit_remaining <= 0:
error = MockError(
error_type=MockErrorType.RATE_LIMIT,
message="Rate limit exceeded. Try again later.",
status_code=429
)
self.add_to_history(request, error.to_openai_format())
raise error
# 에러 시나리오 체크
for error_type, config in self.error_config.items():
if random.random() < config["probability"]:
status_codes = {
MockErrorType.RATE_LIMIT: 429,
MockErrorType.TIMEOUT: 408,
MockErrorType.SERVER_ERROR: 500,
MockErrorType.AUTH_ERROR: 401,
MockErrorType.INVALID_REQUEST: 400,
MockErrorType.NETWORK_ERROR: 503
}
error = MockError(
error_type=error_type,
message=config["message"],
status_code=status_codes.get(error_type, 500)
)
self.add_to_history(request, error.to_openai_format())
raise error
# 정상 응답 생성
self.rate_limit_remaining -= 1
last_message = messages[-1]["content"] if messages else ""
response = {
"id": f"mock-adv-{len(self.call_history)}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": f"[Advanced Mock] Processed: {last_message[:30]}..."
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15
},
"x-ratelimit-remaining": str(self.rate_limit_remaining),
"x-ratelimit-reset": str(int(self.rate_limit_reset or time.time() + 3600))
}
self.add_to_history(request, response)
return response
def get_history(self) -> list:
"""호출 이력 반환"""
return self.call_history
def get_statistics(self) -> dict:
"""통계 정보 반환"""
return {
"total_calls": len(self.call_history),
"rate_limit_remaining": self.rate_limit_remaining,
"configured_errors": list(self.error_config.keys())
}
사용 예시
if __name__ == "__main__":
client = AdvancedMockClient()
# 시나리오 1: 10% 확률로 레이트 리밋 발생
client.configure_error_scenario(
MockErrorType.RATE_LIMIT,
probability=0.1,
message="Rate limit exceeded after 100 requests"
)
# 시나리오 2: 5% 확률로 서버 에러 발생
client.configure_error_scenario(
MockErrorType.SERVER_ERROR,
probability=0.05,
message="Internal server error during processing"
)
# 지연 시간: 평균 200ms (실제 GPT-4 평균 지연 시간)
client.set_latency(200)
# 레이트 리밋: 5회 호출 후 제한
client.set_rate_limit(remaining=5, reset_timestamp=time.time() + 3600)
# 테스트 실행
for i in range(10):
try:
result = client.chat_completions(
model="gpt-4.1",
messages=[{"role": "user", "content": f"Test message {i}"}]
)
print(f"[{i+1}] Success: {result['choices'][0]['message']['content'][:50]}")
print(f" Rate Limit Remaining: {result.get('x-ratelimit-remaining', 'N/A')}")
except MockError as e:
print(f"[{i+1}] Error ({e.error_type.value}): {e.message}")
# 통계 출력
stats = client.get_statistics()
print(f"\n=== Statistics ===")
print(f"Total Calls: {stats['total_calls']}")
print(f"Rate Limit Remaining: {stats['rate_limit_remaining']}")
print(f"Configured Errors: {stats['configured_errors']}")
자주 발생하는 오류와 해결책
오류 1: Mock 응답 형식이 실제 API와 호환되지 않음
# ❌ 잘못된 접근 - 불완전한 Mock 응답
def bad_mock():
return {"result": "Hello World"} # OpenAI 형식 아님
✅ 올바른 접근 - HolySheep AI / OpenAI 호환 형식
def correct_mock():
return {
"id": "chatcmpl-123",
"object": "chat.completion",
"created": 1677652288,
"model": "gpt-4.1",
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello World"
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15
}
}
Stream 응답 Mock도 동일하게 적용
def correct_stream_mock():
return {
"id": "chatcmpl-123",
"object": "chat.completion.chunk",
"created": 1677652288,
"model": "gpt-4.1",
"choices": [{
"index": 0,
"delta": {
"content": "Hello"
},
"finish_reason": None
}]
}
오류 2: 환경 변수 누락으로 프로덕션에서 Mock 모드 유지
import os
import sys
❌ 위험한 접근 - 기본값이 Mock이면 프로덕션에서도 Mock 사용 가능
def bad_client():
env = os.getenv("AI_ENV", "mock") # 잘못된 기본값
if env == "mock":
return MockClient()
return RealClient()
✅ 안전한 접근 - 명시적 환경 설정 강제
def safe_client():
env = os.getenv("AI_API_ENV")
if env is None:
print("⚠️ AI_API_ENV가 설정되지 않았습니다.")
print(" development: Mock API 사용")
print(" production: HolySheep AI API 사용")
sys.exit(1)
clients = {
"development": MockClient,
"testing": MockClient,
"production": lambda: HolySheepAIClient(os.environ["HOLYSHEEP_API_KEY"])
}
client_class = clients.get(env)
if client_class is None:
raise ValueError(f"알 수 없는 환경: {env}. 허용값: {list(clients.keys())}")
return client_class()
.env.example 파일에 반드시 포함해야 할 내용
DOTENV_CONTENT = """
.env.example - 복사하여 .env로 이름 변경 후填写
AI_API_ENV=development # development, testing, production
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY # production에서만 필요
"""
오류 3: 비동기 코드에서 Mock 지연 시간 누락
import asyncio
import time
❌ 비동기 함수에서 time.sleep() 블로킹 발생
async def bad_async_mock():
time.sleep(1) # 이벤트 루프 블로킹!
return {"content": "result"}
✅ asyncio.sleep() 사용 - 논블로킹 지연
async def good_async_mock():
await asyncio.sleep(1) # 다른 태스크 실행 가능
return {"content": "result"}
실제 테스트
async def test_mock_performance():
client = AsyncMockClient()
start = time.time()
tasks = [client.chat_completions(f"request-{i}") for i in range(10)]
results = await asyncio.gather(*tasks) # 동시 실행
elapsed = time.time() - start
# 10개 요청이 1초(지연) 만에 완료되어야 함
assert elapsed < 2.0, f"너무 오래 걸림: {elapsed:.2f}초"
print(f"10개 동시 요청 완료: {elapsed*1000:.0f}ms")
async Mock 클라이언트 예시
class AsyncMockClient:
async def chat_completions(self, request_id: str):
await asyncio.sleep(0.1) # 100ms 지연
return {
"id": request_id,
"choices": [{"message": {"content": f"Response for {request_id}"}}]
}
if __name__ == "__main__":
asyncio.run(test_mock_performance())
오류 4: 레이트 리밋 시뮬레이션 누락
import time
from collections import deque
✅ 레이트 리밋을 포함한 Mock 클라이언트
class RateLimitedMockClient:
def __init__(self, rpm: int = 60, tpm: int = 100000):
self.rpm = rpm
self.tpm = tpm
self.request_timestamps: deque = deque(maxlen=rpm)
self.total_tokens_used: int = 0
self.window_start: float = time.time()
def _check_rate_limit(self, estimated_tokens: int):
current_time = time.time()
# RPM 체크 (1분 윈도우)
while self.request_timestamps and \
current_time - self.request_timestamps[0] > 60:
self.request_timestamps.popleft()
if len(self.request_timestamps) >= self.rpm:
wait_time = 60 - (current_time - self.request_timestamps[0])
raise Exception(f"RPM_limit_exceeded: {len(self.request_timestamps)}/{self.rpm}. Wait {wait_time:.1f}s")
# TPM 체크 (1분 윈도우)
if self.total_tokens_used + estimated_tokens > self.tpm:
if current_time - self.window_start >= 60:
self.total_tokens_used = 0
self.window_start = current_time
else:
raise Exception(f"TPM_limit_exceeded: {self.total_tokens_used}/{self.tpm}")
self.request_timestamps.append(current_time)
self.total_tokens_used += estimated_tokens
def chat_completions(self, messages: list, **kwargs):
estimated_tokens = sum(len(str(m)) for m in messages) // 4
self._check_rate_limit(estimated_tokens)
return {
"id": f"mock-{len(self.request_timestamps)}",
"usage": {
"prompt_tokens": estimated_tokens,
"completion_tokens": 20,
"total_tokens": estimated_tokens + 20
}
}
테스트
client = RateLimitedMockClient(rpm=5) # 분당 5회 제한
for i in range(7):
try:
result = client.chat_completions([{"role": "user", "content":