본문으로 건너뛰기

© 2026 Molayo

Dev.to헤드라인2026. 05. 12. 23:23

Novastack을 사용한 Python 기반 다중 모델 AI 라우터 구축 🚀

요약

본 기사는 여러 LLM 제공업체(DeepSeek, Claude, Qwen 등)를 사용하는 프로덕션급 AI 애플리케이션의 유지보수 문제를 다루고 있습니다. Novastack의 토큰 포워딩 플랫폼을 활용하여 단일 OpenAI 호환 엔드포인트와 하나의 API 키로 여러 모델에 접근하는 '다중 모델 라우터' 구축 방법을 제시합니다. 이 라우터를 사용하면 작업 유형(코드 리뷰, 창작, 분석 등)에 따라 가장 적합한 모델을 자동으로 선택하고, 복잡한 장애 조치 로직 없이 안정적인 AI 서비스를 구현할 수 있습니다.

핵심 포인트

  • 다중 LLM 관리의 어려움: 여러 API 키, 클라이언트 라이브러리, 복잡한 라우팅 로직 관리가 필요함.
  • Novastack 솔루션: 단일 OpenAI 호환 엔드포인트를 통해 여러 모델에 대한 접근을 통합하여 유지보수성을 극대화함.
  • 자동 라우팅 구현: 사용자의 요청(task) 내용을 분석하여 코드 생성, 창작, 분석 등 작업 유형별로 최적의 모델을 자동으로 선택할 수 있음.
  • 안정적인 아키텍처 구축: 단일 엔드포인트와 폴백 로직을 통해 서비스 안정성을 높이고 복잡한 장애 조치 코드를 제거함.

프로덕션급 AI 애플리케이션을 개발할 때, 여러 LLM(대규모 언어 모델) 제공업체에 대한 여러 API 키, 엔드포인트 및 클라이언트 라이브러리를 관리하는 것은 유지보수 측면에서 악몽이 됩니다. 오늘 저는 Novastack의 토큰 포워딩 플랫폼을 사용하여 단일 엔드포인트를 통해 세 가지 강력한 모델에 대한 접근 방식을 통합하는 방법을 보여드리겠습니다.

왜 다중 모델 라우터가 중요한가? 가장 큰 이점은 AI 애플리케이션이 다양한 작업을 위해 다른 모델을 사용할 수 있다는 점입니다. 예를 들어, DeepSeek-V4-Pro는 코드 생성을 처리하고 Claude-Opus-4.7은 창의적인 작문을 관리할 수 있습니다. 하지만 기존 방식으로는 다음 사항들이 필요합니다:

  • 각 제공업체별 별도의 API 키
  • 서로 다른 클라이언트 라이브러리 및 인증 방법
  • 제공업체가 다운되었을 때의 수동 장애 조치(failover) 로직
  • 코드베이스 전체에 흩어져 있는 복잡한 라우팅 로직

Novastack은 사용자가 지정하는 모델 매개변수에 따라 요청을 적절한 모델로 라우팅하는 단일 OpenAI 호환 엔드포인트를 제공함으로써 이를 우아하게 해결합니다. 하나의 키, 하나의 엔드포인트, 세 개의 모델입니다.

라우터 설정하기

깨끗한 오류 처리와 폴백(fallback) 기능을 유지하면서 요청을 지능적으로 안내하는 프로덕션급 모델 라우터를 구축해 봅시다.

필수 조건:

pip install openai httpx

https://novapai.ai에서 API 키를 받은 다음, 환경을 설정합니다:

import os
from typing import Optional,
                    Dict, 
                    Any, 
                    List
from openai import OpenAI # Novastack 구성
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 =

AI 자동 생성 콘텐츠

본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.

원문 바로가기
3

댓글

0