LemonData logoLemonData

설정

언어
블로그

하나의 API 키로 AI 챗봇 만들기: 30분 만에 기초부터 배포까지

L
LemonData
·2026년 2월 26일·11 조회수
#챗봇#튜토리얼#파이썬#FastAPI#스트리밍
하나의 API 키로 AI 챗봇 만들기: 30분 만에 기초부터 배포까지

하나의 API 키로 AI 챗봇 만들기: 30분 만에 제로에서 프로덕션까지

이 튜토리얼에서는 스트리밍 응답, 대화 기록, 모델 전환, 적절한 오류 처리를 갖춘 프로덕션 준비 완료 AI 챗봇 백엔드를 구축합니다. Python, FastAPI, OpenAI SDK를 사용하며 API 집계기를 통해 어떤 모델이든 사용할 수 있습니다.

필수 조건

pip install fastapi uvicorn openai

1단계: 기본 채팅 엔드포인트

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-lemon-xxx",
    base_url="https://api.lemondata.cc/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "gpt-4.1-mini"
    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단계: 스트리밍 추가

스트리밍은 전체 응답을 기다리지 않고 토큰이 생성되는 즉시 전송합니다. 사용자는 실시간으로 답변이 형성되는 것을 볼 수 있습니다.

@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"
    )

3단계: 대화 기록

대화 기록을 메모리에 저장합니다(프로덕션에서는 Redis나 데이터베이스로 교체하세요).

from collections import defaultdict
import uuid

conversations: dict[str, list] = defaultdict(list)

SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    conv_id = req.conversation_id or str(uuid.uuid4())

    # 메시지 기록 구성
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend(conversations[conv_id])
    messages.append({"role": "user", "content": req.message})

    # 사용자 메시지 저장
    conversations[conv_id].append(
        {"role": "user", "content": req.message}
    )

    def generate():
        full_response = []
        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"

        # 어시스턴트 응답 저장
        conversations[conv_id].append(
            {"role": "assistant", "content": "".join(full_response)}
        )
        yield f"data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"X-Conversation-ID": conv_id}
    )

4단계: 오류 처리

AI API 호출은 여러 이유로 실패할 수 있습니다: 속도 제한, 잔액 부족, 모델 미사용 가능 등. 각 경우를 처리하세요:

from openai import (
    APIError,
    RateLimitError,
    APIConnectionError
)

@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():
        try:
            full_response = []
            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"

            conversations[conv_id].append(
                {"role": "assistant", "content": "".join(full_response)}
            )

        except RateLimitError as e:
            yield f"data: [ERROR] 속도 제한에 걸렸습니다. 잠시 기다려 주세요.\n\n"
        except APIConnectionError:
            yield f"data: [ERROR] 연결 실패. 재시도 중...\n\n"
        except APIError as e:
            yield f"data: [ERROR] {e.message}\n\n"

        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    # 컨텍스트 길이 관리를 위해 최근 10턴 유지
    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

5단계: 모델 전환

사용자가 대화 중간에 모델을 전환할 수 있도록 합니다. 다양한 필요에 맞는 모델들:

AVAILABLE_MODELS = {
    "fast": "gpt-4.1-mini",
    "smart": "claude-sonnet-4-6",
    "reasoning": "o3",
    "budget": "deepseek-chat",
    "creative": "claude-sonnet-4-6",
}

@app.get("/models")
async def list_models():
    return {"models": AVAILABLE_MODELS}

프론트엔드에서 이 옵션들을 제공할 수 있습니다. 모든 모델이 집계기를 통해 동일한 OpenAI 호환 형식을 사용하므로, 전환은 model 파라미터만 변경하면 됩니다.

6단계: 컨텍스트 윈도우 관리

긴 대화는 모델 컨텍스트 한도를 초과할 수 있습니다. 슬라이딩 윈도우를 구현하세요:

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    """시스템 프롬프트 + 최근 메시지를 토큰 예산 내에 유지합니다."""
    # 대략적인 추정: 1 토큰 ≈ 4 문자
    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

완성된 애플리케이션

# 실행: uvicorn main:app --reload --port 8000
# 테스트: curl -N -X POST http://localhost:8000/chat/stream \
#   -H "Content-Type: application/json" \
#   -d '{"message": "Hello!", "model": "gpt-4.1-mini"}'

전체 코드는 100줄 미만입니다. 여기서 추가할 수 있는 기능:

  • 인증 (API 키 또는 JWT)
  • 영속적 저장소 (대화용 PostgreSQL 또는 Redis)
  • 사용자별 속도 제한
  • 사용량 추적 및 청구
  • 양방향 스트리밍을 위한 WebSocket 지원
  • 프론트엔드 (React, Vue 또는 EventSource를 사용하는 순수 JS)

비용 추정

하루 1,000건 대화(평균 5턴) 처리하는 챗봇 기준:

모델 일일 비용 월간 비용
GPT-4.1-mini 약 $2.40 약 $72
GPT-4.1 약 $12.00 약 $360
Claude Sonnet 4.6 약 $18.00 약 $540
DeepSeek V3 약 $1.68 약 $50

대부분 대화는 GPT-4.1-mini를 사용하고, 사용자가 요청할 때만 Claude Sonnet 4.6으로 업그레이드하면 대부분 애플리케이션에서 월 $100 이하로 비용을 유지할 수 있습니다.

API 키 받기: lemondata.cc는 하나의 엔드포인트로 300개 이상의 모델을 제공합니다. $1 무료 크레딧으로 시작하세요.

Share: