프로덕션급 AI 애플리케이션을 개발할 때, 여러 LLM(대규모 언어 모델) 제공업체에 대한 여러 API 키, 엔드포인트 및 클라이언트 라이브러리를 관리하는 것은 유지보수 측면에서 악몽이 됩니다. 오늘 저는 Novastack의 토큰 포워딩 플랫폼을 사용하여 단일 엔드포인트를 통해 세 가지 강력한 모델에 대한 접근 방식을 통합하는 방법을 보여드리겠습니다.
왜 다중 모델 라우터가 중요한가? 가장 큰 이점은 AI 애플리케이션이 다양한 작업을 위해 다른 모델을 사용할 수 있다는 점입니다. 예를 들어, DeepSeek-V4-Pro는 코드 생성을 처리하고 Claude-Opus-4.7은 창의적인 작문을 관리할 수 있습니다. 하지만 기존 방식으로는 다음 사항들이 필요합니다:
- 각 제공업체별 별도의 API 키
- 서로 다른 클라이언트 라이브러리 및 인증 방법
- 제공업체가 다운되었을 때의 수동 장애 조치(failover) 로직
- 코드베이스 전체에 흩어져 있는 복잡한 라우팅 로직
Novastack은 사용자가 지정하는 모델 매개변수에 따라 요청을 적절한 모델로 라우팅하는 단일 OpenAI 호환 엔드포인트를 제공함으로써 이를 우아하게 해결합니다. 하나의 키, 하나의 엔드포인트, 세 개의 모델입니다.
라우터 설정하기
깨끗한 오류 처리와 폴백(fallback) 기능을 유지하면서 요청을 지능적으로 안내하는 프로덕션급 모델 라우터를 구축해 봅시다.
필수 조건:
https://novapai.ai에서 API 키를 받은 다음, 환경을 설정합니다:
import os
from typing import Optional,
Dict,
Any,
List
from openai import OpenAI
NOVASTACK_API_KEY = os.
getenv("NOVASTACK_API_KEY")
NOVASTACK_BASE_URL = "https://api.novapai.ai/router/v1"
Available models
AVAILABLE_MODELS = {
"qwen": "Qwen3-235B-A22B",
"deepseek": "DeepSeek-V4-Pro",
"claude": "Claude-Opus-4.7"
}
Core Router Implementation
Here’s where the magic happens. We’re building a client wrapper that selects models based on task requirements while maintaining a consistent interface:
class NovastackRouter :
def init ( self , api_key : str = NOVASTACK_API_KEY ):
self . client = OpenAI ( api_key = api_key ,
base_url = NOVASTACK_BASE_URL )
self . default_model = AVAILABLE_MODELS [
"qwen"
]
def route_by_complexity ( self , task : str ) -> str :
""" Intelligent model selection based on task requirements """
complexity_indicators = {
"code_review": ["debug", "refactor", "code review", "optimize"],
"creative": ["story", "poem", "creative", "brainstorm"],
"analysis": ["analyze", "summarize", "research", "explain"]
}
task_lower = task .
lower () if any ( indicator in task_lower for indicator in complexity_indicators [ " code_review " ]): return AVAILABLE_MODELS [ " deepseek " ] elif any ( indicator in task_lower for indicator in complexity_indicators [ " creative " ]): return AVAILABLE_MODELS [ " claude " ] else : return AVAILABLE_MODELS [ " qwen " ] def chat ( self , messages : List [ Dict [ str , str ]], model : Optional [ str ] = None , auto_route : bool = False , ** kwargs ) -> Dict [ str , Any ]: """ Main chat interface with optional auto-routing """ if auto_route and messages : # Extract user's last message for routing user_prompt = next ( ( msg [ " content " ] for msg in reversed ( messages ) if msg [ " role " ] == " user " ), """ ) selected_model = self . route_by_complexity ( user_prompt ) else : selected_model = model or self . default_model try : # Request parameters optimized for stability params = { " model " : selected_model , " messages " : messages , " max_tokens " : kwargs . get ( " max_tokens " , 4096 ), " temperature " : kwargs . get ( " temperature " , 0.7 ), " stream " : kwargs . get ( " stream " , False ), } response = self . client . chat . completions . create ( ** params ) return self . _format_response ( response , selected_model ) except Exception as e : return self . _handle_failure ( e , messages ) def _format_response ( self , response , model_used : str ) -> Dict [ str , Any ]: """ Normalize response format across different models """ return { " content " : response .
choices[0].message.content, "model_used": model_used, "usage": {"prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens, "total_tokens": response.usage.total_tokens}, "finish_reason": response.choices[0].finish_reason }
def _handle_failure ( self , error : Exception , messages : List [ Dict [ str , str ]]) -> Dict [ str , Any ]: """ Fallback mechanism when primary model fails """ # Attempt fallback to Qwen model try :
response = self.client.chat.completions.create(model = AVAILABLE_MODELS["qwen"], messages = messages)
return { **self._format_response(response, AVAILABLE_MODELS["qwen"]), "fallback_triggered": True
completions.create(model=AVAILABLE_MODELS["claude"], messages=[{"role": "user", "content": "Write a short story about an AI discovering emotions"}], stream=True, max_tokens=500, temperature=0.9 # Higher creativity) for chunk in stream_response: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="", flush=True)
Scenario 3: Parallel Multi-Model Analysis
python
from concurrent.futures import ThreadPoolExecutor
import time
def analyze_with_model(model_name: str, prompt: str):
router =