안녕하세요, 저는 HolySheep AI의 시니어 엔지니어입니다. 이번 튜토리얼에서는 LangChain의 스트리밍(Streaming) 출력机制을 깊이 파고들고, 이를 프론트엔드와 실시간 통합하는 프로덕션 레벨 아키텍처를 다루겠습니다. API 호출 지연 시간부터 동시성 제어, 비용 최적화까지 실전에서 검증된 전략을 공유합니다.
왜 스트리밍이 중요한가?
AI 응답의 체감 지연 시간은用户体验의 핵심 지표입니다. 스트리밍 없이 30초 소요되는 응답을 스트리밍 적용 시 TTFT(Time To First Token)를 500ms 이하로 줄일 수 있어, Claude Sonnet 4.5나 GPT-4.1 같은 고가 모델에서 사용자 이탈률을 크게 낮출 수 있습니다.
HolySheep AI 게이트웨이(지금 가입)는 스트리밍 응답에 최적화된 인프라를 제공하며, DeepSeek V3.2의 경우 $0.42/MTok라는 혁신적인 가격으로 스트리밍 기반 애플리케이션의 운영 비용을 획기적으로 절감할 수 있습니다.
아키텍처 설계
스트리밍 아키텍처의 핵심은 이벤트 기반 파이프라인입니다. 다음은 전체 데이터 플로우입니다:
┌─────────────┐ ┌─────────────────┐ ┌──────────────┐ ┌─────────────┐
│ Frontend │────▶│ FastAPI/BFF │────▶│ LangChain │────▶│ HolySheep │
│ (SSE/WSS) │◀────│ (Streaming) │◀────│ Callback │◀────│ AI Gateway │
└─────────────┘ └─────────────────┘ └──────────────┘ └─────────────┘
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────────┐ ┌──────────────┐
│ 사용자 화면 │ │ 버퍼링/재시도 │ │ 토큰 카운팅 │
│ 실시간 갱신 │ │ 에러 처리 │ │ 비용 추적 │
└─────────────┘ └─────────────────┘ └──────────────┘1. LangChain 스트리밍 기본 구현
1.1 커스텀 Callback Handler 만들기
LangChain의 스트리밍 핵심은 CallbackHandler 인터페이스입니다. 저는 실제 프로덕션 환경에서 검증된 StreamingCallbackHandler를 구현하겠습니다.
import asyncio
import time
from typing import Any, Dict, List, Optional
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from dataclasses import dataclass, field
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
@dataclass
class TokenMetrics:
"""토큰 메트릭 수집용 데이터 클래스"""
tokens_per_second: float = 0.0
total_tokens: int = 0
first_token_latency_ms: float = 0.0
completion_time_ms: float = 0.0
start_time: Optional[float] = None
first_token_time: Optional[float] = None
class StreamingCallbackHandler(BaseCallbackHandler):
"""
HolySheep AI 스트리밍 응답용 커스텀 콜백 핸들러
주요 기능:
- 실시간 토큰 스트리밍
- TTFT(Time To First Token) 측정
- 토큰 처리량 모니터링
- 비용 추적 (HolySheep AI 가격 정책 적용)
"""
def __init__(
self,
token_buffer: asyncio.Queue,
model_name: str = "gpt-4.1",
api_provider: str = "holysheep"
):
self.token_buffer = token_buffer
self.metrics = TokenMetrics()
self.model_name = model_name
self.api_provider = api_provider
self._tokens_received = []
# HolySheep AI 가격 정책 (2024년 기준, USD per 1M tokens)
self.pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4": 15.0, # $15/MTok
"gemini-2.5-flash": 2.5, # $2.50/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok
}
def on_llm_start(
self,
serialized: Dict[str, Any],
prompts: List[str],
**kwargs: Any
) -> None:
"""LLM 호출 시작 시 실행"""
self.metrics = TokenMetrics()
self.metrics.start_time = time.perf_counter()
self._tokens_received = []
logger.info(f"[LLM Start] Model: {self.model_name}, Prompt tokens: ~{len(prompts[0])}")
def on_llm_new_token(
self,
token: str,
*,
chunk: Optional[Any] = None,
**kwargs: Any
) -> None:
"""새 토큰 수신 시 실행 - 스트리밍 핵심"""
current_time = time.perf_counter()
# TTFT 측정
if self.metrics.first_token_time is None:
self.metrics.first_token_time = current_time
if self.metrics.start_time:
self.metrics.first_token_latency_ms = (
current_time - self.metrics.start_time
) * 1000
logger.info(
f"[TTFT] {self.metrics.first_token_latency_ms:.2f}ms"
)
# 토큰 버퍼에 추가
self._tokens_received.append(token)
self.metrics.total_tokens = len(self._tokens_received)
# 버퍼가 가득차면 비동기로 전달
if len(token) > 0:
asyncio.create_task(self.token_buffer.put({
"token": token,
"timestamp": current_time,
"total_so_far": self.metrics.total_tokens
}))
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
"""LLM 호출 완료 시 실행"""
if self.metrics.start_time:
self.metrics.completion_time_ms = (
time.perf_counter() - self.metrics.start_time
) * 1000
if self.metrics.completion_time_ms > 0:
self.metrics.tokens_per_second = (
self.metrics.total_tokens /
(self.metrics.completion_time_ms / 1000)
)
# 비용 계산
cost_usd = self._calculate_cost()
logger.info(
f"[LLM End] Total tokens: {self.metrics.total_tokens}, "
f"Time: {self.metrics.completion_time_ms:.2f}ms, "
f"TPS: {self.metrics.tokens_per_second:.2f}, "
f"Cost: ${cost_usd:.6f}"
)
def _calculate_cost(self) -> float:
"""HolySheep AI 가격 정책 기반 비용 계산"""
price_per_mtok = self.pricing.get(
self.model_name,
self.pricing["gpt-4.1"]
)
# 출력 토큰만 계산 (입력 토큰은 프로ンプ트에 따라 다름)
return (self.metrics.total_tokens / 1_000_000) * price_per_mtok
def get_metrics(self) -> TokenMetrics:
"""수집된 메트릭 반환"""
return self.metrics1.2 HolySheep AI 게이트웨이 연동
이제 HolySheep AI의 LangChain 통합을 설정합니다. 반드시 https://api.holysheep.ai/v1을 base_url로 사용해야 합니다.
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio
class HolySheepStreamingChain:
"""
HolySheep AI 스트리밍 체인
HolySheep AI의 단일 API 키로 여러 모델 지원:
- GPT-4.1: $8/MTok (고품질 작업)
- Claude Sonnet 4.5: $15/MTok (복잡한 추론)
- Gemini 2.5 Flash: $2.50/MTok (대량 처리/비용 최적화)
- DeepSeek V3.2: $0.42/MTok ( budget-friendly)
"""
def __init__(
self,
api_key: str = None,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048
):
# HolySheep AI 필수 설정
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
self.model = model
self.temperature = temperature
self.max_tokens = max_tokens
# 모델별 최적 설정
self.model_configs = {
"gpt-4.1": {"response_format": "text"},
"claude-sonnet-4": {"thinking": {"type: "enabled", "budget_tokens": 1024}},
"gemini-2.5-flash": {"response_mime_type": "text/plain"},
"deepseek-v3.2": {}
}
self._llm = None
def _get_llm(self) -> ChatOpenAI:
"""지연 초기화로 LLM 인스턴스 생성"""
if self._llm is None:
self._llm = ChatOpenAI(
model=self.model,
api_key=self.api_key,
base_url=self.base_url,
temperature=self.temperature,
max_tokens=self.max_tokens,
streaming=True, # 스트리밍 활성화
**self.model_configs.get(self.model, {})
)
return self._llm
async def astream_with_callback(
self,
prompt: str,
token_buffer: asyncio.Queue,
system_prompt: str = None
):
"""
스트리밍 출력 + 콜백 핸들러
Args:
prompt: 사용자 입력 프롬프트
token_buffer: 토큰 수신용 asyncio 큐
system_prompt: 시스템 프롬프트 (선택)
Yields:
각 토큰을 실시간으로 yield
"""
from main import StreamingCallbackHandler # 이전에 정의한 핸들러
# 프롬프트 구성
messages = []
if system_prompt:
messages.append(("system", system_prompt))
messages.append(("human", prompt))
prompt_template = ChatPromptTemplate.from_messages(messages)
chain = prompt_template | self._get_llm() | StrOutputParser()
# 콜백 핸들러 설정
callback_handler = StreamingCallbackHandler(
token_buffer=token_buffer,
model_name=self.model
)
# 스트리밍 실행
full_response = ""
async for token in chain.astream(
input={},
config={"callbacks": [callback_handler]}
):
full_response += token
yield token
# 최종 메트릭 반환
return callback_handler.get_metrics()
사용 예시
async def main():
chain = HolySheepStreamingChain(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2" # 비용 최적화 모델
)
token_queue = asyncio.Queue()
print("Starting streaming response...\n")
async for token in chain.astream_with_callback(
prompt="AI의 미래에 대해 500단어로 설명해줘",
token_buffer=token_queue
):
print(token, end="", flush=True)
# 메트릭 확인
metrics = await token_queue.get()
print(f"\n\n[Metrics] Total: {metrics.total_tokens} tokens")
if __name__ == "__main__":
asyncio.run(main())2. FastAPI 백엔드 통합 (SSE)
프론트엔드와 실시간 통신하려면 Server-Sent Events(SSE)가 가장 효율적입니다. WebSocket보다 구현이 간단하고 HTTP/2环境下에서 최적의 성능을 발휘합니다.
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import Optional, AsyncGenerator
import asyncio
import json
import logging
from datetime import datetime
from main import HolySheepStreamingChain, StreamingCallbackHandler
from langchain_core.callbacks import BaseCallbackHandler
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="HolySheep AI Streaming API")
CORS 설정 (프론트엔드 도메인에 맞게 조정)
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://your-domain.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class ChatRequest(BaseModel):
"""채팅 요청 스키마"""
message: str = Field(..., min_length=1, max_length=10000)
system_prompt: Optional[str] = Field(None, max_length=2000)
model: str = Field(default="deepseek-v3.2")
temperature: float = Field(default=0.7, ge=0, le=2.0)
max_tokens: int = Field(default=2048, ge=1, le=8192)
session_id: Optional[str] = None
class TokenEvent:
"""SSE 이벤트 포맷터"""
@staticmethod
def format(token: str, event_type: str = "token") -> bytes:
"""SSE 형식으로 변환"""
data = {
"type": event_type,
"content": token,
"timestamp": datetime.utcnow().isoformat()
}
return f"data: {json.dumps(data, ensure_ascii=False)}\n\n".encode("utf-8")
@staticmethod
def format_metrics(metrics) -> bytes:
"""메트릭 데이터 포맷"""
data = {
"type": "metrics",
"total_tokens": metrics.total_tokens,
"ttft_ms": round(metrics.first_token_latency_ms, 2),
"tps": round(metrics.tokens_per_second, 2),
"total_time_ms": round(metrics.completion_time_ms, 2)
}
return f"event: metrics\ndata: {json.dumps(data)}\n\n".encode("utf-8")
@staticmethod
def format_error(error: str) -> bytes:
"""에러 포맷"""
data = {
"type": "error",
"message": error
}
return f"event: error\ndata: {json.dumps(data)}\n\n".encode("utf-8")
@staticmethod
def heartbeat() -> bytes:
"""연결 유지용 하트비트"""
return b": heartbeat\n\n"
class StreamAggCallback(BaseCallbackHandler):
"""
SSE 스트리밍을 위한 LangChain 콜백 핸들러
토큰을 수집하여 SSE 포맷으로 변환
"""
def __init__(self, queue: asyncio.Queue):
self.queue = queue
async def on_llm_new_token(self, token: str, **kwargs):
await self.queue.put(("token", token))
async def generate_sse_stream(
chain: HolySheepStreamingChain,
request: ChatRequest
) -> AsyncGenerator[bytes, None]:
"""
SSE 스트리밍 응답 생성기
HolySheep AI 게이트웨이를 통한 실시간 토큰 스트리밍을
SSE 형식으로 프론트엔드에 전달
"""
token_queue: asyncio.Queue[dict] = asyncio.Queue()
callback = StreamingCallbackHandler(
token_buffer=token_queue,
model_name=request.model
)
# 하트비트 간격 (연결 유지)
heartbeat_interval = 15 # 15초
async def token_collector():
"""백그라운드 토큰 수집 태스크"""
try:
async for token in chain.astream_with_callback(
prompt=request.message,
token_buffer=token_queue,
system_prompt=request.system_prompt
):
pass # 토큰은 queue에 이미 저장됨
except Exception as e:
logger.error(f"Chain execution error: {e}")
await token_queue.put(("error", str(e)))
# 토큰 수집 시작
collector_task = asyncio.create_task(token_collector())
try:
last_heartbeat = datetime.utcnow()
while True:
try:
# 토큰 또는 이벤트 대기 (최대 30초)
item = await asyncio.wait_for(
token_queue.get(),
timeout=30.0
)
if item[0] == "error":
yield TokenEvent.format_error(item[1])
break
# 토큰 이벤트 전송
if item[0] == "token":
yield TokenEvent.format(item[1])
# 메트릭 확인
if item.get("total_so_far", 0) > 0:
metrics = callback.get_metrics()
# 100토큰마다 메트릭 전송
if item["total_so_far"] % 100 == 0:
yield TokenEvent.format_metrics(metrics)
last_heartbeat = datetime.utcnow()
except asyncio.TimeoutError:
# 타임아웃 시 하트비트 전송
now = datetime.utcnow()
if (now - last_heartbeat).seconds >= heartbeat_interval:
yield TokenEvent.heartbeat()
last_heartbeat = now
except asyncio.CancelledError:
logger.info("SSE stream cancelled")
finally:
collector_task.cancel()
try:
await collector_task
except asyncio.CancelledError:
pass
# 최종 메트릭 전송
metrics = callback.get_metrics()
yield TokenEvent.format_metrics(metrics)
# 스트림 종료 신호
yield b"event: done\ndata: {}\n\n"
@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
"""
스트리밍 채팅 엔드포인트
POST /api/chat/stream
Request Body:
{
"message": "안녕하세요, AI에 대해 알려주세요",
"model": "deepseek-v3.2",
"temperature": 0.7
}
Response: Server-Sent Events (SSE)
"""
# 모델 유효성 검사
valid_models = ["gpt-4.1", "claude-sonnet-4", "gemini-2.5-flash", "deepseek-v3.2"]
if request.model not in valid_models:
raise HTTPException(
status_code=400,
detail=f"Invalid model. Choose from: {valid_models}"
)
chain = HolySheepStreamingChain(
model=request.model,
temperature=request.temperature,
max_tokens=request.max_tokens
)
return StreamingResponse(
generate_sse_stream(chain, request),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # Nginx 버퍼링 비활성화
}
)
@app.get("/health")
async def health_check():
"""헬스체크 엔드포인트"""
return {"status": "healthy", "provider": "HolySheep AI"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)3. 프론트엔드 통합 (React + SSE)
import React, { useState, useCallback, useRef, useEffect } from 'react';
interface StreamMessage {
type: 'token' | 'metrics' | 'error' | 'done';
content?: string;
total_tokens?: number;
ttft_ms?: number;
tps?: number;
total_time_ms?: number;
message?: string;
}
interface UseStreamingChatOptions {
apiUrl?: string;
apiKey?: string;
model?: string;
onComplete?: (metrics: StreamMessage) => void;
onError?: (error: Error) => void;
}
interface UseStreamingChatReturn {
messages: Array<{role: 'user' | 'assistant'; content: string}>;
isStreaming: boolean;
metrics: StreamMessage | null;
sendMessage: (message: string, systemPrompt?: string) => Promise;
error: Error | null;
}
export function useStreamingChat(
options: UseStreamingChatOptions = {}
): UseStreamingChatReturn {
const {
apiUrl = 'http://localhost:8000/api/chat/stream',
apiKey,
model = 'deepseek-v3.2',
onComplete,
onError
} = options;
const [messages, setMessages] = useState<
Array<{role: 'user' | 'assistant'; content: string}>
>([]);
const [isStreaming, setIsStreaming] = useState(false);
const [metrics, setMetrics] = useState(null);
const [error, setError] = useState(null);
const abortControllerRef = useRef(null);
const currentResponseRef = useRef('');
const sendMessage = useCallback(async (
message: string,
systemPrompt?: string
) => {
// 이전 요청 취소
if (abortControllerRef.current) {
abortControllerRef.current.abort();
}
// 새 AbortController 생성
abortControllerRef.current = new AbortController();
// 사용자 메시지 추가
setMessages(prev => [...prev, { role: 'user', content: message }]);
setIsStreaming(true);
setError(null);
currentResponseRef.current = '';
try {
const response = await fetch(apiUrl, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
// HolySheep AI API 키 (필요시)
...(apiKey && { 'Authorization': Bearer ${apiKey} })
},
body: JSON.stringify({
message,
system_prompt: systemPrompt,
model,
temperature: 0.7,
max_tokens: 2048
}),
signal: abortControllerRef.current.signal
});
if (!response.ok) {
throw new Error(HTTP ${response.status}: ${response.statusText});
}
const reader = response.body?.getReader();
if (!reader) {
throw new Error('Response body is not readable');
}
const decoder = new TextDecoder();
let buffer = '';
let finalMetrics: StreamMessage | null = null;
// SSE 파싱 및 실시간 렌더링
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() || '';
for (const line of lines) {
if (line.startsWith('event: ')) {
const eventType = line.slice(7).trim();
// 다음 줄에서 데이터 읽기
const dataLineIndex = lines.indexOf(line) + 1;
if (dataLineIndex < lines.length) {
const dataLine = lines[dataLineIndex];
if (dataLine.startsWith('data: ')) {
try {
const data = JSON.parse(dataLine.slice(6));
if (eventType === 'token' || data.type === 'token') {
// 토큰 실시간 추가
currentResponseRef.current += data.content || data;
setMessages(prev => {
const lastMsg = prev[prev.length - 1];
if (lastMsg?.role === 'assistant') {
return [
...prev.slice(0, -1),
{ ...lastMsg, content: currentResponseRef.current }
];
}
return [...prev, { role: 'assistant', content: data.content || data }];
});
} else if (eventType === 'metrics') {
finalMetrics = data;
setMetrics(data);
} else if (eventType === 'error') {
throw new Error(data.message || 'Unknown error');
} else if (eventType === 'done') {
// 스트리밍 완료
}
} catch (parseError) {
// JSON 파싱 실패는 무시 (빈 줄 등)
}
}
}
} else if (line.startsWith('data: ') && !line.includes('event:')) {
// event 없이 data만 오는 경우
try {
const data = JSON.parse(line.slice(6));
if (data.type === 'token' || data.content) {
currentResponseRef.current += data.content || '';
setMessages(prev => {
const lastMsg = prev[prev.length - 1];
if (lastMsg?.role === 'assistant') {
return [
...prev.slice(0, -1),
{ ...lastMsg, content: currentResponseRef.current }
];
}
return [...prev, { role: 'assistant', content: data.content || '' }];
});
}
} catch {
// 토큰 데이터
if (!line.includes('event:')) {
currentResponseRef.current += line.replace('data: ', '');
setMessages(prev => {
const lastMsg = prev[prev.length - 1];
if (lastMsg?.role === 'assistant') {
return [
...prev.slice(0, -1),
{ ...lastMsg, content: currentResponseRef.current }
];
}
return [...prev, { role: 'assistant', content: line.replace('data: ', '') }];
});
}
}
}
}
}
// 완료 콜백
if (finalMetrics && onComplete) {
onComplete(finalMetrics);
}
} catch (err) {
if ((err as Error).name !== 'AbortError') {
const error = err as Error;
setError(error);
if (onError) onError(error);
}
} finally {
setIsStreaming(false);
}
}, [apiUrl, apiKey, model, onComplete, onError]);
// 컴포넌트 언마운트 시 요청 취소
useEffect(() => {
return () => {
if (abortControllerRef.current) {
abortControllerRef.current.abort();
}
};
}, []);
return { messages, isStreaming, metrics, sendMessage, error };
}
// 사용 예시 컴포넌트
export function ChatComponent() {
const { messages, isStreaming, metrics, sendMessage, error } = useStreamingChat({
model: 'deepseek-v3.2',
onComplete: (metrics) => {
console.log('Streaming complete:', metrics);
}
});
const [input, setInput] = useState('');
const handleSubmit = (e: React.FormEvent) => {
e.preventDefault();
if (input.trim() && !isStreaming) {
sendMessage(input);
setInput('');
}
};
return (
{/* 메시지 목록 */}
{messages.map((msg, i) => (
message ${msg.role}}>
{msg.role}
{msg.content}
))}
{isStreaming && (
AI가 입력 중...
)}
{/* 메트릭 표시 */}
{metrics && (
토큰: {metrics.total_tokens}
TTFT: {metrics.ttft_ms}ms
TPS: {metrics.tps?.toFixed(2)}
)}
{/* 에러 표시 */}
{error && (
오류: {error.message}
)}
{/* 입력 폼 */}
);
} 성능 최적화 전략
3.1 동시성 제어
다중 사용자 환경에서 API 호출 동시성을 제어하지 않으면 HolySheep AI의 Rate Limit에 도달하거나 비용이 급증할 수 있습니다. 저는 asyncio.Semaphore와 Redis 기반 분산 락을 조합하여 프로덕션 레벨의 동시성 제어를 구현합니다.
import asyncio
from typing import Dict, Optional
from datetime import datetime, timedelta
import time
from collections import defaultdict
import threading
class RateLimiter:
"""
HolySheep AI API를 위한 동시성 및 Rate Limit 제어기
HolySheep AI 모델별 Rate Limit:
- GPT-4.1: 500 RPM / 80,000 TPM
- Claude Sonnet 4.5: 300 RPM / 200,000 TPM
- Gemini 2.5 Flash: 1000 RPM / 1,000,000 TPM
- DeepSeek V3.2: 2000 RPM / 1,000,000 TPM
"""
def __init__(
self,
rpm_limit: int = 500,
tpm_limit: int = 80000,
concurrent_limit: int = 10
):
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
self.concurrent_limit = concurrent_limit
# Rate Limiters
self._rpm_semaphore = asyncio.Semaphore(concurrent_limit)
self._request_timestamps: Dict[str, list] = defaultdict(list)
self._token_counts: Dict[str, list] = defaultdict(list)
self._lock = asyncio.Lock()
# 모니터링
self._stats = {
"total_requests": 0,
"total_tokens": 0,
"rejected_requests": 0,
"avg_latency_ms": 0
}
async def acquire(
self,
client_id: str,
estimated_tokens: int = 500
) -> Optional[float]:
"""
Rate Limit 내에서 실행 허용 여부 확인
Returns:
대기 시간 (초) 또는 None (즉시 실행 가능)
"""
async with self._lock:
now = time.time()
window_start = now - 60 # 1분 윈도우
# 윈도우 내 요청 수 확인
self._request_timestamps[client_id] = [
ts for ts in self._request_timestamps[client_id]
if ts > window_start
]
# RPM 체크
if len(self._request_timestamps[client_id]) >= self.rpm_limit:
oldest_request = min(self._request_timestamps[client_id])
wait_time = 60 - (now - oldest_request)
return wait_time if wait_time > 0 else 0
# TPM 체크
self._token_counts[client_id] = [
(ts, tokens) for ts, tokens in self._token_counts[client_id]
if ts > window_start
]
current_tpm = sum(
tokens for _, tokens in self._token_counts[client_id]
)
if current_tpm + estimated_tokens > self.tpm_limit:
oldest_token = min(
ts for ts, _ in self._token_counts[client_id]
)
wait_time = 60 - (now - oldest_token)
return wait_time if wait_time > 0 else 0
# 요청 허용
self._request_timestamps[client_id].append(now)
self._token_counts[client_id].append((now, estimated_tokens))
return None
async def execute_with_limit(
self,
client_id: str,
coro,
estimated_tokens: int = 500
):
"""Rate Limit 내에서 코루틴 실행"""
# 대기 시간 확인
wait_time = await self.acquire(client_id, estimated_tokens)
if wait_time:
self._stats["rejected_requests"] += 1
raise RateLimitError(
f"Rate limit reached. Retry after {wait_time:.2f} seconds"
)
# 동시성 제한 내에서 실행
start_time = time.perf_counter()
async with self._rpm_semaphore:
try:
result = await coro
# 성공 메트릭 업데이트
latency_ms = (time.perf_counter() - start_time) * 1000
self._update_stats(estimated_tokens, latency_ms)
return result
except Exception as e:
# 실패 시 통계 업데이트
self._stats["rejected_requests"] += 1
raise
def _update_stats(self, tokens: int, latency_ms: float):
"""통계 업데이트"""
self._stats["total_requests"] += 1
self._stats["total_tokens"] += tokens
# 이동 평균으로 지연 시간 업데이트
n = self._stats["total_requests"]
current_avg = self._stats["avg_latency_ms"]
self._stats["avg_latency_ms"] = (
(current_avg * (n - 1) + latency_ms) / n
)
def get_stats(self) -> Dict:
"""통계 반환"""
return {
**self._stats,
"active_requests": self._rpm_semaphore._value,
"timestamp": datetime.utcnow().isoformat()
}
class RateLimitError(Exception):
"""Rate Limit 초과 에러"""
pass
모델별 Rate Limiter 인스턴스
rate_limiters = {
"gpt-4.1": RateLimiter(rpm_limit=500, tpm_limit=80000, concurrent_limit=10),
"claude-sonnet-4": RateLimiter(rpm_limit=300, tpm_limit=200000, concurrent_limit