Written with AI assistance from the source video. The video is the authority.
9:38Better Stack
Log in to leave a comment
No posts yet
DeepSeek Chat, GPT-4o-mini, Gemini 2.5 Flash 같은 저가형 모델을 쓰면 API 비용은 확실히 줄어듭니다. 문제는 트래픽이 조금만 몰려도 응답 시간이 10초에서 60초까지 치솟는다는 점입니다. 일반적인 동기식 HTTP 요청 구조에서 이런 지연이 발생하면 서버 워커 프로세스가 대기 시간에 묶여 먹통이 되고, 프론트엔드는 타임아웃 에러를 뱉으며 뻗어버립니다.
혼자서 사이드 프로젝트를 만들거나 예산이 빠듯한 1인 개발자에게 서버 마비는 치명적입니다. 그렇다고 비싼 모델로 돌아갈 수는 없습니다. 클라이언트와 LLM 간의 직접적인 동기 커넥션을 끊고, 모델의 조악한 출력을 프론트엔드에 안전하게 전달하는 비동기 분리 아키텍처를 직접 구축해야 합니다.
웹 서버가 LLM의 추론 완료를 직접 기다리게 만들면 안 됩니다. FastAPI를 API 게이트웨이로 두고, Redis 메시지 브로커와 Celery 분산 워커를 결합해 요청 접수와 실제 추론 연산을 완전히 분리해야 합니다. 사용자가 요청을 보내면 서버는 작업 ID만 즉시 반환하고 연결을 끊습니다.
기본 설정 그대로 Celery를 돌리면 워커가 터졌을 때 작업이 유실되거나 중복 실행됩니다. LLM 추론 특성상 길어지는 작업 시간을 소화하려면 visibility_timeout을 3600초로 늘리고, 작업이 정상 종료된 시점에만 수신 확인을 보내도록 task_acks_late = True를 지정해야 합니다. 워커 하나가 여러 헤비한 작업을 한 번에 쥐고 병목을 일으키지 않도록 worker_prefetch_multiplier = 1을 세팅하는 것도 필수입니다.
# config.py
from celery import Celery
REDIS_URL = "redis://localhost:6379/0"
celery_app = Celery("ai_tasks", broker=REDIS_URL, backend=REDIS_URL)
celery_app.conf.update(
broker_transport_options={"visibility_timeout": 3600},
task_acks_late=True,
task_reject_on_worker_lost=True,
worker_prefetch_multiplier=1,
task_soft_time_limit=180,
task_time_limit=240,
task_serializer="json",
result_serializer="json",
accept_content=["json"],
)
백그라운드에서 실행할 LLM 호출 함수에는 지수 백오프 재시도 로직을 겁니다. 네트워크 찰나의 오류나 순간적인 429 에러로 작업 전체가 실패하는 불상사를 막아줍니다.
# tasks.py
import requests
from config import celery_app
@celery_app.task(bind=True, name="tasks.generate_ai_response")
def generate_ai_response(self, prompt: str):
try:
response = requests.post(
"https://api.deepseek.com/v1/chat/completions",
json={
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3
},
headers={"Authorization": "Bearer YOUR_API_KEY"},
timeout=150
)
response.raise_for_status()
data = response.json()
return {"status": "SUCCESS", "result": data["choices"][0]["message"]["content"]}
except Exception as exc:
raise self.retry(exc=exc, countdown=2 ** self.request.retries, max_retries=3)
FastAPI 엔드포인트는 요청이 오자마자 Celery 큐에 할일을 던지고 202 Accepted 응답을 내보냅니다. 프론트엔드는 이 작업 ID를 받아 periodic polling 방식으로 상태를 확인합니다.
# main.py
from fastapi import FastAPI
from pydantic import BaseModel
from celery.result import AsyncResult
from tasks import generate_ai_response
app = FastAPI()
class PromptRequest(BaseModel):
prompt: str
@app.post("/api/v1/tasks/generate", status_code=202)
async def create_task(payload: PromptRequest):
task = generate_ai_response.delay(payload.prompt)
return {"job_id": task.id, "status": "PENDING", "poll_url": f"/api/v1/tasks/{task.id}"}
@app.get("/api/v1/tasks/{job_id}")
async def get_task_status(job_id: str):
task_result = AsyncResult(job_id)
if task_result.state == "PENDING":
return {"job_id": job_id, "status": "PENDING"}
elif task_result.state == "SUCCESS":
return {"job_id": job_id, "status": "SUCCESS", "data": task_result.result}
elif task_result.state == "FAILURE":
return {"job_id": job_id, "status": "FAILED", "error": str(task_result.info)}
return {"job_id": job_id, "status": task_result.state}
이 구조를 잡고 나면 API 게이트웨이의 웹 워커가 LLM 응답을 기다리느라 블로킹되는 현상이 사라집니다. 클라이언트에게는 진행 상황을 폴링으로 보여주면 그만입니다.
저렴한 모델일수록 HTML이나 inline CSS를 직접 렌더링하라고 시키면 맥을 못 춥니다. 닫는 태그를 누락하거나 문법을 깨뜨려 프론트엔드 레이아웃 전체를 붕괴시킵니다. 모델에게 마크다운이나 코드 블록을 절대 뱉지 못하게 하고, 딱 필요한 데이터만 정형화된 JSON 형태로 받아서 프론트엔드의 사전에 정의된 컴포넌트에 꽂아 넣어야 합니다.
Pydantic으로 허용할 UI 데이터 타입을 정의하고, 모델 출력값이 이 규칙을 위반하면 에러 알림용 컴포넌트 Props로 안전하게 전환합니다.
# ui_schema.py
import re
import json
from typing import List, Optional, Literal, Union
from pydantic import BaseModel, ValidationError
class CardProps(BaseModel):
title: str
description: str
badge_text: Optional[str] = None
action_label: str
class AlertProps(BaseModel):
variant: Literal["info", "success", "warning", "error"]
title: str
message: str
class TableProps(BaseModel):
headers: List[str]
rows: List[List[str]]
class UIResponse(BaseModel):
component_type: Literal["CARD", "ALERT", "TABLE"]
props: Union[CardProps, AlertProps, TableProps]
UI_SYSTEM_PROMPT = """
You are a strict JSON UI generator. Respond ONLY with a single valid JSON object matching the schema below.
DO NOT include markdown formatting (like ```json), commentary, or extra HTML tags.
Allowed Component Types:
1. "CARD": props = {"title": str, "description": str, "badge_text": str (optional), "action_label": str}
2. "ALERT": props = {"variant": "info"|"success"|"warning"|"error", "title": str, "message": str}
3. "TABLE": props = {"headers": [str], "rows": [[str]]}
"""
def parse_llm_ui_response(raw_response: str) -> UIResponse:
json_match = re.search(r'\{.*\}', raw_response, re.DOTALL)
if not json_match:
return UIResponse(
component_type="ALERT",
props=AlertProps(variant="error", title="Parsing Error", message="No JSON structure found.")
)
try:
data = json.loads(json_match.group(0))
return UIResponse(**data)
except (json.JSONDecodeError, ValidationError):
return UIResponse(
component_type="ALERT",
props=AlertProps(variant="error", title="Schema Error", message="Invalid props structure.")
)
프론트엔드는 백엔드가 검증해서 넘겨준 component_type을 보고 어떤 Tailwind UI를 그릴지 결정합니다. 모델이 코드를 이상하게 짜서 화면이 하얗게 뜨는 일은 이제 생기지 않습니다.
// UIComponentRenderer.tsx
import React from 'react';
interface ComponentProps {
component_type: 'CARD' | 'ALERT' | 'TABLE';
props: any;
}
export const UIComponentRenderer: React.FC<ComponentProps> = ({ component_type, props }) => {
switch (component_type) {
case 'CARD':
return (
<div className="max-w-sm p-6 bg-white border border-gray-200 rounded-lg shadow-md">
{props.badge_text && (
<span className="bg-blue-100 text-blue-800 text-xs font-semibold px-2.5 py-0.5 rounded">
{props.badge_text}
</span>
)}
<h5 className="mt-2 mb-2 text-2xl font-bold text-gray-900">{props.title}</h5>
<p className="mb-4 font-normal text-gray-700">{props.description}</p>
<button className="px-4 py-2 text-sm font-medium text-white bg-blue-700 rounded-lg">
{props.action_label}
</button>
</div>
);
case 'ALERT':
const styles = {
info: 'text-blue-800 bg-blue-50 border-blue-300',
success: 'text-green-800 bg-green-50 border-green-300',
warning: 'text-yellow-800 bg-yellow-50 border-yellow-300',
error: 'text-red-800 bg-red-50 border-red-300',
};
return (
<div className={`p-4 border rounded-lg ${styles[props.variant as keyof typeof styles]}`} role="alert">
<span className="font-bold">{props.title}:</span> {props.message}
</div>
);
case 'TABLE':
return (
<div className="relative overflow-x-auto shadow-md sm:rounded-lg">
<table className="w-full text-sm text-left text-gray-500">
<thead className="text-xs text-gray-700 uppercase bg-gray-50">
<tr>
{props.headers.map((h: string, idx: number) => (
<th key={idx} className="px-6 py-3">{h}</th>
))}
</tr>
</thead>
<tbody>
{props.rows.map((row: string[], rIdx: number) => (
<tr key={rIdx} className="bg-white border-b">
{row.map((cell: string, cIdx: number) => (
<td key={cIdx} className="px-6 py-4">{cell}</td>
))}
</tr>
))}
</tbody>
</table>
</div>
);
default:
return null;
}
};
저렴한 모델이라고 다 같은 비용 구조를 갖는 것은 아닙니다. RAG나 에이전트 서비스처럼 프롬프트 길이가 길어지는 작업에서는 입력 토큰과 출력 토큰 비율이 보통 4:1 이상으로 벌어집니다. 표기된 입력 단가만 볼 게 아니라 실제 혼합 단가를 계산해두고 우선순위를 매겨야 합니다.
| 모델 명칭 | 제공사 | 입력 단가 ($/1M) | 출력 단가 (/1M, 4:1) |
|---|---|---|---|
| DeepSeek Chat (V3) | DeepSeek | $0.14 | $0.28 |
| GPT-4o-mini | OpenAI | $0.15 | $0.60 |
| Gemini 2.5 Flash | $0.10 | $0.40 | |
| Claude 3 Haiku | Anthropic | $0.25 | $1.25 |
단가가 가장 낮은 모델을 primary로 두고, 해당 API에 장애가 나거나 지연 시간이 설정한 서킷 브레이커 기준을 넘어서면 차선책 모델로 즉시 우회시켜야 합니다. LiteLLM Router 라이브러리를 쓰면 코드 몇 줄로 이 로직을 구현할 수 있습니다.
# fallback_router.py
import os
import logging
from litellm import Router
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("LLMRouter")
model_list = [
{
"model_name": "primary-budget-model",
"litellm_params": {
"model": "deepseek/deepseek-chat",
"api_key": os.getenv("DEEPSEEK_API_KEY"),
"timeout": 10,
},
},
{
"model_name": "fallback-secondary-model",
"litellm_params": {
"model": "gpt-4o-mini",
"api_key": os.getenv("OPENAI_API_KEY"),
"timeout": 15,
},
},
{
"model_name": "fallback-tertiary-model",
"litellm_params": {
"model": "gemini/gemini-2.5-flash",
"api_key": os.getenv("GEMINI_API_KEY"),
"timeout": 20,
},
}
]
router = Router(
model_list=model_list,
fallbacks=[
{"primary-budget-model": ["fallback-secondary-model", "fallback-tertiary-model"]}
],
num_retries=1,
cooldown_time=300,
)
async def execute_llm_completion(prompt: str):
try:
response = await router.acompletion(
model="primary-budget-model",
messages=[{"role": "user", "content": prompt}],
temperature=0.2
)
logger.info(f"Success using [{response.model}]. Tokens: {response.usage.total_tokens}")
return {
"content": response.choices[0].message.content,
"model_used": response.model,
"tokens": response.usage.total_tokens
}
except Exception as e:
logger.error(f"All fallbacks failed: {str(e)}")
raise RuntimeError("Service unavailable due to upstream failures.")
비동기 작업 큐로 백엔드를 분리하고, Pydantic 검증으로 프론트엔드를 보호하며, LiteLLM 라우터로 API 장애를 방어하는 조합입니다. 저렴한 모델의 고질적인 불안정성을 감싸안으면서도 서비스 가용성을 일정 수준 이상으로 유지하는 현실적인 방안입니다.