이 튜토리얼에서는 FastAPI, SSE 스트리밍, 대화 메모리, 모델 전환 기능을 갖춘 작지만 프로덕션 수준의 챗봇 서비스를 구축합니다. 단순히 데모를 만드는 것이 목표가 아닙니다. 실제 제품 서비스 뒤에 배치하여 모델을 변경할 때마다 통합 코드를 다시 작성할 필요 없이 안전하게 반복 개선할 수 있는 백엔드를 만드는 것이 목표입니다.
이미 OpenAI 호환 SDK를 TokenLab에 연결했다면 이 글부터 시작하면 됩니다. 아직 기본 URL을 변경하지 않았다면 먼저 마이그레이션 가이드를 읽어보세요. 부하 발생 시 요청 조정 및 백오프(backoff)가 주된 관심사라면 이 가이드와 함께 AI API 속도 제한 가이드를 참고하세요.
핵심 요약
- 프로덕션 수준의 챗봇에는 동기식 엔드포인트, 스트리밍 엔드포인트, 서버 측 대화 상태, 모델 허용 목록(allowlist), 실제 오류 처리, 명확한 스토리지 업그레이드 경로라는 6가지 요소가 필요합니다.
- 스트리밍, 메모리, 도구 호출을 추가하기 전에 하나의 작은 채팅 엔드포인트로 API 키, 기본 URL, 라우팅이 정상 작동하는지 먼저 확인하세요.
- SSE 스트리밍은 대부분의 채팅 제품을 커버하며 웹소켓보다 운영 오버헤드가 적습니다.
- 프론트엔드에서 임의의 모델 ID나 더 이상 사용되지 않는 모델 ID를 요청할 수 없도록 자유 텍스트 필드가 아닌 백엔드 허용 목록을 통해 모델을 노출하세요.
- 모델 가용성과 라인업은 자주 변경됩니다. 허용 목록을 프로덕션에 고정하기 전에 TokenLab의 모델 디렉토리(2026-07-07 확인 기준)를 확인하세요.
구축할 서비스
완성된 서비스는 다음과 같은 6가지 구성 요소로 이루어집니다:
- 스모크 테스트를 위한 동기식
/chat엔드포인트. - 실제 UI를 위한 스트리밍
/chat/stream엔드포인트. conversation_id로 구분되는 대화 상태.- 프론트엔드에서 임의의 ID를 요청하지 못하도록 하는 모델 허용 목록.
- 첫 번째 429 오류에서 멈추지 않는 오류 처리.
- 인메모리 프로토타입에서 Redis나 PostgreSQL로 넘어가는 명확한 경로.
이 정도면 지원 봇, 내부 어시스턴트 또는 임베디드 채팅 위젯의 첫 번째 버전을 구동하기에 충분합니다.
최소 스택 설치
pip install fastapi uvicorn openai pydantic redis
첫 단계에서는 redis를 생략할 수 있지만, 지금 미리 임포트를 설정해두면 나중에 리팩토링할 필요 없이 쉽게 업그레이드할 수 있습니다.
1단계: 작고 단순한 채팅 엔드포인트로 시작하기
챗봇 작업에서 길을 잃는 가장 빠른 방법은 기본적인 요청 경로가 안정화되기도 전에 웹소켓, 도구 사용, 에이전트 오케스트레이션을 시작하는 것입니다. 키, 기본 URL, 모델 라우팅이 작동함을 증명하는 작은 엔드포인트 하나로 시작하세요.
from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-tokenlab-xxx",
base_url="https://api.tokenlab.sh/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "deepseek-v4-flash"
conversation_id: str | None = None
@app.post("/chat")
async def chat(req: ChatRequest):
response = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}]
)
return {"reply": response.choices[0].message.content}
스모크 테스트를 한 번 실행해보세요. 여기서 실패한다면 그 위에 계속 빌드하지 마세요.
2단계: 사용자가 측정하기 전에 지연 시간을 느끼므로 스트리밍 추가하기
대부분의 챗봇 제품이 느리게 느껴지는 이유는 모델이 느려서가 아니라 전체 응답이 도착할 때까지 UI가 비어 있기 때문입니다. SSE는 대부분의 채팅 제품에 충분하며 웹소켓보다 운영 부담이 적습니다.
from fastapi.responses import StreamingResponse
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
def generate():
stream = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
yield f"data: {delta.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
프론트엔드에서는 일반적인 fetch reader만으로도 충분합니다:
async function sendMessage(payload) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
console.log(chunk);
}
}
제품이 이미 표준 HTTP 기반의 브라우저 클라이언트를 사용 중이라면, SSE가 웹소켓 계층보다 아키텍처를 더 단순하게 유지해줍니다.
3단계: 대화 상태를 요청 본문에서 분리하기
첫 번째 챗봇 데모는 보통 전체 대화 기록을 브라우저에 보관하고 매번 다시 전송합니다. 프로토타입에는 효과적이지만, 재시도, 세션 재개, 서버 측 도구 사용이 필요해지면 금방 복잡해집니다.
시작 단계에서는 인메모리 저장소로도 충분합니다:
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
history = conversations[conv_id][-20:]
messages.extend(history)
messages.append({"role": "user", "content": user_msg})
conversations[conv_id].append({"role": "user", "content": user_msg})
return messages
Redis로 업그레이드하는 과정은 로직 변경보다는 스토리지 연결 작업에 가깝습니다:
import json
import redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def load_history(conv_id: str) -> list:
raw = redis_client.get(f"chat:{conv_id}")
return json.loads(raw) if raw else []
def save_history(conv_id: str, history: list) -> None:
redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))
대화에 TTL(유효 시간), 재개 가능성, 다중 인스턴스 배포가 필요할 때 Redis를 사용하세요. 대화 기록 자체가 나중에 쿼리, 감사, 보고해야 할 제품 데이터라면 PostgreSQL을 사용하세요.
4단계: 오류를 예외가 아닌 제품 동작으로 처리하기
고객 대상 챗봇이라면 성공 경로만큼이나 실패 경로도 중요합니다. 사용자는 실패 원인이 속도 제한인지, 잔액 부족인지, 상위 모델의 중단인지 알 필요가 없습니다. UI가 멈추는지 여부가 중요합니다.
from openai import APIConnectionError, APIError, RateLimitError
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
def generate():
full_response = []
try:
stream = client.chat.completions.create(
model=req.model,
messages=messages,
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response.append(delta.content)
yield f"data: {delta.content}\n\n"
except RateLimitError:
yield "data: [error: rate limited, please retry shortly]\n\n"
except APIConnectionError:
yield "data: [error: connection issue, please retry]\n\n"
except APIError:
yield "data: [error: something went wrong on our end]\n\n"
finally:
if full_response:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
5단계: 프론트엔드에서 요청 가능한 모델 제한하기
프론트엔드가 임의의 모델 문자열을 API로 직접 전달하게 하지 마세요. 자유 텍스트 필드는 더 이상 사용되지 않는 모델, 오타, 또는 노출할 의도가 없는 모델에 대한 요청을 유발합니다. 대신 백엔드 허용 목록을 통해 라우팅하세요.
ALLOWED_MODELS = {
"default": "deepseek-v4-flash",
"flagship": "gpt-5.5",
"balanced": "claude-sonnet-5",
}
class ChatRequest(BaseModel):
message: str
tier: str = "default"
conversation_id: str | None = None
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
# 나머지 스트리밍 로직은 클라이언트가 제공한 원시 문자열 대신 `model` 변수를 사용
이렇게 하면 공급자가 모델을 중단했을 때 프론트엔드 코드를 수정하거나 클라이언트 업데이트를 배포할 필요 없이 한 곳에서 모델을 교체할 수 있습니다.
6단계: 성공 경로뿐만 아니라 프로덕션의 나머지 부분 처리하기
챗봇 백엔드는 핵심 채팅 호출이 똑똑해질 때가 아니라, 주변의 엣지 케이스가 처리될 때 프로덕션 준비가 완료된 것으로 간주됩니다.
체크리스트는 간단합니다:
- 프론트엔드 실패를 백엔드 로그와 연결할 수 있도록 요청 ID 추가
- 사용자당 동시성 및 요청 크기 제한
- 토큰 예산을 초과하기 전에 긴 대화 기록 자르기
- 모델, 지연 시간, 입력 크기, 종료 이유 기록
- 사용자에게 보이는 오류 메시지와 내부 오류 세부 정보 분리
- 첫 번째 실제 중단이 발생하기 전에 대체 모델 하나를 테스트하여 장애 복구 작동 확인
기록 자르기는 간단하게 유지할 수 있습니다:
def trim_history(messages: list, max_tokens: int = 8000) -> list:
system = messages[0]
history = messages[1:]
total_chars = len(system["content"])
trimmed = []
for msg in reversed(history):
msg_chars = len(msg["content"])
if total_chars + msg_chars > max_tokens * 4:
break
trimmed.insert(0, msg)
total_chars += msg_chars
return [system] + trimmed
핵심은 토큰을 완벽하게 계산하는 것이 아니라, 비용이나 지연 시간에 영향을 미치기 전에 명백한 컨텍스트 폭발을 막는 것입니다.
데모에서 제품으로
이 백엔드가 안정화되면 다음 업그레이드는 보통 "더 많은 AI"가 아닙니다. 보통은 지루한 인프라 작업입니다:
- 한 사용자가 다른 사용자의 대화를 읽을 수 없도록 하는 인증
- 배포 후에도 세션이 유지되도록 하는 영속성
- 시끄러운 사용자가 할당량을 소진하지 못하도록 하는 속도 제한
- 챗봇이 고객 대상일 경우의 과금 또는 사용량 귀속
- 대화에 장기 기억이 필요할 경우의 백그라운드 요약
통합 게이트웨이는 이 작업의 대부분을 도와줍니다. 기본 URL 마이그레이션을 마치면, 모델 변경은 플랫폼 재작성이 아니라 설정 편집이 됩니다.
스모크 테스트
uvicorn main:app --reload --port 8000
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "model": "deepseek-v4-flash"}'
한 번의 턴을 스트리밍하고, 대화 하나를 보존하며, 강제 실패 시 깔끔한 오류를 반환할 수 있다면 올바른 기반을 갖춘 것입니다.
비용 추정
TokenLab에서 API 키를 생성하고 OpenAI SDK를 https://api.tokenlab.sh/v1로 지정하면, 공급자별로 별도의 계정을 관리할 필요 없이 챗봇의 첫 번째 프로덕션 버전을 배포할 수 있습니다.
| 모델 | 일반적인 티어 | 참고 |
|---|---|---|
| DeepSeek V4 Flash | 빠름 / 기본 | 대용량, 저지연 턴에 적합한 기본값 |
| GPT-5.5 | 플래그십 | 더 강력한 추론이 필요한 턴에 사용 |
| Claude Sonnet 5 | 균형 | 코딩 및 검토 스타일의 답변에 강력 추천 |
| Gemini 3.5 Flash | 예산 / 빠른 대안 | 대용량 라우팅을 위한 저비용, 빠른 대안 |
정확한 토큰당 가격은 공급자마다 자주 변경되므로 여기에 고정된 수치로 기재하지 않았습니다. 예산을 책정하기 전에 모델 디렉토리(2026-07-07 확인 기준)에서 현재 요금을 확인하세요. 실제로는 대부분의 대화를 DeepSeek V4 Flash나 Gemini 3.5 Flash 같은 빠름/기본 티어로 라우팅하고, 필요한 경우에만 GPT-5.5나 Claude Sonnet 5를 예약하여 사용하면 대부분의 애플리케이션에서 월 비용을 낮게 유지할 수 있습니다. 하지만 예산을 확정하기 전에 계정의 실제 백만 토큰당 요금을 확인하세요.
FAQ
AI 챗봇을 만들려면 웹소켓이 필요한가요? 아니요. 2단계에서 보여드린 SSE 스트리밍으로 대부분의 채팅 제품을 커버할 수 있습니다. 웹소켓은 라이브 협업이나 서버 시작 이벤트처럼 요청/응답 외의 양방향 푸시가 필요할 때 진정한 가치를 발휘합니다. 표준 채팅 UI의 경우 SSE가 배포, 디버깅, 확장이 더 간단합니다.
기본 모델을 무엇으로 설정해야 할까요? 기본 티어로는 DeepSeek V4 Flash나 Gemini 3.5 Flash 같은 빠르고 저렴한 모델로 시작하고, 5단계에서 보여드린 허용 목록 뒤에 Claude Sonnet 5나 GPT-5.5를 사용한 균형 잡힌 또는 추론 티어를 추가하세요. 새로운 모델이 출시되고 기존 모델이 중단되는 일정이 통제 범위를 벗어나므로 모델 디렉토리(2026-07-07 확인 기준)에서 현재 옵션을 확인하세요.
챗봇이 데모에서 실제 트래픽으로 넘어갈 때 가장 먼저 고장 나는 것은 무엇인가요? 거의 항상 성공 경로가 아니라 오류 경로입니다. 무제한 재시도, 사용자당 동시성 제한 누락, 무제한 대화 기록은 실제 부하 상태에서 챗봇 백엔드가 무너지는 가장 흔한 세 가지 원인입니다. 위의 4단계와 6단계에서 이 세 가지를 직접 다룹니다.
API 키로 시작하기: tokenlab.sh는 하나의 엔드포인트를 통해 300개 이상의 모델을 제공합니다. 구축을 시작할 수 있도록 $1의 무료 크레딧을 드립니다.
출처
2026-07-07 기준 가격
- TokenLab model directory2026-07-07 기준 확인
