AI 앱 풀스택 개발 가이드: FastAPI + Next.js로 LLM 서비스 만들기

개요

현대 AI 애플리케이션 개발은 단순한 API 호출을 넘어 스트리밍, 멀티모달, RAG(Retrieval-Augmented Generation) 등 복잡한 기술을 통합해야 합니다. 이 가이드는 FastAPI 백엔드와 Next.js 프론트엔드를 기반으로 프로덕션 수준의 LLM 서비스를 구축하는 전 과정을 다룹니다.

1. AI 앱 아키텍처 설계

현대 AI 앱의 3계층 구조

현대적인 AI 애플리케이션은 세 가지 핵심 레이어로 구성됩니다.

프론트엔드 레이어: Next.js App Router, Vercel AI SDK, 스트리밍 UI
백엔드 레이어: FastAPI, LangChain, 인증 미들웨어, 캐싱
AI/데이터 레이어: OpenAI/Claude API, 벡터 데이터베이스, 임베딩 모델

스트리밍 vs 배치 처리

LLM 응답을 처리하는 두 가지 주요 방식이 있습니다.

스트리밍 처리는 토큰이 생성되는 즉시 클라이언트로 전송하는 방식입니다. 사용자 체감 응답 속도가 빠르고 대화형 인터페이스에 적합합니다. Server-Sent Events(SSE) 또는 WebSocket을 사용합니다.

배치 처리는 전체 응답이 완성된 후 한 번에 반환하는 방식입니다. 문서 처리, 데이터 분석, 배경 작업에 적합합니다. Celery + Redis 큐를 활용합니다.

프로젝트 폴더 구조

ai-app/
├── backend/
│   ├── app/
│   │   ├── main.py
│   │   ├── routers/
│   │   │   ├── chat.py
│   │   │   └── documents.py
│   │   ├── services/
│   │   │   ├── llm_service.py
│   │   │   └── vector_service.py
│   │   └── models/
│   │       └── schemas.py
│   ├── requirements.txt
│   └── Dockerfile
├── frontend/
│   ├── app/
│   │   ├── chat/
│   │   │   └── page.tsx
│   │   └── api/
│   │       └── chat/
│   │           └── route.ts
│   ├── components/
│   └── package.json
└── docker-compose.yml

2. FastAPI 백엔드 구성

설치 및 기본 설정

pip install fastapi uvicorn openai langchain langchain-openai python-dotenv

Pydantic 모델로 요청/응답 검증

# app/models/schemas.py
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class Role(str, Enum):
    user = "user"
    assistant = "assistant"
    system = "system"

class Message(BaseModel):
    role: Role
    content: str

class ChatRequest(BaseModel):
    messages: List[Message]
    model: str = Field(default="gpt-4o-mini")
    temperature: float = Field(default=0.7, ge=0, le=2)
    max_tokens: Optional[int] = Field(default=None)

class ChatResponse(BaseModel):
    content: str
    usage: dict

비동기 스트리밍 엔드포인트

FastAPI의 StreamingResponse를 사용하면 LLM 토큰을 실시간으로 클라이언트에 전송할 수 있습니다.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from openai import AsyncOpenAI
from app.models.schemas import ChatRequest

app = FastAPI(title="AI App Backend")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["*"],
    allow_headers=["*"],
)

client = AsyncOpenAI()

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        stream = await client.chat.completions.create(
            model=request.model,
            messages=[m.dict() for m in request.messages],
            stream=True,
            temperature=request.temperature,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {delta}\n\n"
        yield "data: [DONE]\n\n"

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

의존성 주입 패턴

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="토큰이 만료되었습니다."
        )

@app.post("/api/chat/secure")
async def secure_chat(request: ChatRequest, user=Depends(verify_token)):
    # 인증된 사용자만 접근 가능
    pass

3. LangChain 통합

대화 체인 구성

LangChain을 사용하면 메모리 관리, 체인 구성, 도구 통합을 쉽게 할 수 있습니다.

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
memory = ConversationBufferWindowMemory(k=10)

template = """당신은 친절한 AI 어시스턴트입니다.
현재 대화:
{history}
Human: {input}
AI:"""

prompt = PromptTemplate(
    input_variables=["history", "input"],
    template=template
)

chain = ConversationChain(llm=llm, memory=memory, prompt=prompt)

response = chain.predict(input="안녕하세요, 저는 Python 개발자입니다.")

RAG 파이프라인 구축

RAG(검색 증강 생성)는 외부 문서를 검색하여 LLM의 응답 품질을 높이는 패턴입니다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader

# 문서 로드 및 청킹
loader = PyPDFLoader("document.pdf")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# RAG 체인 구성
llm = ChatOpenAI(model="gpt-4o")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

answer = qa_chain.invoke({"query": "문서의 주요 내용은?"})

커스텀 툴 만들기

from langchain.tools import tool
from langchain.agents import initialize_agent, AgentType

@tool
def search_database(query: str) -> str:
    """데이터베이스에서 정보를 검색합니다. query는 검색할 키워드입니다."""
    # 실제 DB 쿼리 로직
    results = db.search(query)
    return str(results)

@tool
def get_weather(city: str) -> str:
    """특정 도시의 현재 날씨를 조회합니다."""
    response = requests.get(f"https://api.weather.com/v1/{city}")
    return response.json()["description"]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_database, get_weather]
agent = initialize_agent(tools, llm, agent=AgentType.OPENAI_FUNCTIONS)

4. Next.js 프론트엔드

Vercel AI SDK로 스트리밍 채팅

Vercel AI SDK는 Next.js에서 AI 스트리밍을 간편하게 구현하는 공식 라이브러리입니다.

npm install ai @ai-sdk/openai react-markdown

// app/api/chat/route.ts
import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'

export async function POST(req: Request) {
  const { messages } = await req.json()

  const result = await streamText({
    model: openai('gpt-4o-mini'),
    messages,
    system: '당신은 친절하고 도움이 되는 AI 어시스턴트입니다.',
  })

  return result.toDataStreamResponse()
}

채팅 UI 컴포넌트

// app/chat/page.tsx
'use client'
import { useChat } from 'ai/react'
import ReactMarkdown from 'react-markdown'

export default function ChatPage() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } = useChat({
    api: '/api/chat',
  })

  return (
    <div className="flex flex-col h-screen max-w-2xl mx-auto">
      <header className="p-4 border-b font-semibold text-lg">
        AI 어시스턴트
      </header>
      <div className="flex-1 overflow-y-auto p-4 space-y-4">
        {messages.map(m => (
          <div
            key={m.id}
            className={`flex ${m.role === 'user' ? 'justify-end' : 'justify-start'}`}
          >
            <div
              className={`max-w-xs rounded-lg p-3 ${
                m.role === 'user'
                  ? 'bg-blue-500 text-white'
                  : 'bg-gray-100 text-gray-800'
              }`}
            >
              <ReactMarkdown>{m.content}</ReactMarkdown>
            </div>
          </div>
        ))}
        {isLoading && (
          <div className="flex justify-start">
            <div className="bg-gray-100 rounded-lg p-3 text-gray-500">
              답변 생성 중...
            </div>
          </div>
        )}
      </div>
      <form onSubmit={handleSubmit} className="p-4 border-t flex gap-2">
        <input
          value={input}
          onChange={handleInputChange}
          className="flex-1 border rounded-lg px-3 py-2 focus:outline-none focus:ring-2 focus:ring-blue-500"
          placeholder="메시지를 입력하세요..."
          disabled={isLoading}
        />
        <button
          type="submit"
          disabled={isLoading}
          className="bg-blue-500 text-white px-4 py-2 rounded-lg disabled:opacity-50"
        >
          전송
        </button>
      </form>
    </div>
  )
}

파일 업로드 처리

// app/upload/page.tsx
'use client'
import { useState } from 'react'

export default function UploadPage() {
  const [status, setStatus] = useState('')

  async function handleUpload(e: React.FormEvent<HTMLFormElement>) {
    e.preventDefault()
    const formData = new FormData(e.currentTarget)
    setStatus('업로드 중...')

    const response = await fetch('/api/upload', {
      method: 'POST',
      body: formData,
    })

    if (response.ok) {
      const data = await response.json()
      setStatus(`완료: ${data.message}`)
    } else {
      setStatus('업로드 실패')
    }
  }

  return (
    <form onSubmit={handleUpload} className="p-4">
      <input type="file" name="file" accept=".pdf,.txt,.md" />
      <button type="submit" className="mt-2 bg-green-500 text-white px-4 py-2 rounded">
        업로드
      </button>
      {status && <p className="mt-2 text-sm">{status}</p>}
    </form>
  )
}

5. 벡터 데이터베이스 연동

pgvector (PostgreSQL 확장)

PostgreSQL의 pgvector 확장을 사용하면 기존 데이터베이스에서 벡터 검색을 수행할 수 있습니다.

-- pgvector 확장 활성화
CREATE EXTENSION vector;

-- 임베딩 컬럼이 포함된 테이블 생성
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536),
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- HNSW 인덱스 생성 (빠른 근사 최근접 이웃 검색)
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

# pgvector 사용 예
import asyncpg
import numpy as np

async def store_embedding(content: str, embedding: list):
    conn = await asyncpg.connect(DATABASE_URL)
    await conn.execute(
        "INSERT INTO documents (content, embedding) VALUES ($1, $2)",
        content, embedding
    )

async def search_similar(query_embedding: list, k: int = 5):
    conn = await asyncpg.connect(DATABASE_URL)
    results = await conn.fetch(
        """SELECT content, 1 - (embedding <=> $1) as similarity
           FROM documents
           ORDER BY embedding <=> $1
           LIMIT $2""",
        query_embedding, k
    )
    return results

Chroma DB (로컬 개발용)

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="my_documents",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# 문서 추가
vectorstore.add_texts(
    texts=["Python은 AI 개발에 널리 사용됩니다.", "FastAPI는 고성능 API 프레임워크입니다."],
    metadatas=[{"source": "intro.txt"}, {"source": "framework.txt"}]
)

# 유사도 검색
results = vectorstore.similarity_search("API 개발", k=3)

6. 인증 및 보안

JWT 토큰 인증

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get("sub")
    if username is None:
        raise HTTPException(status_code=401, detail="유효하지 않은 토큰")
    return username

프롬프트 인젝션 방어

import re

INJECTION_PATTERNS = [
    r"ignore previous instructions",
    r"disregard all prior",
    r"you are now",
    r"act as",
    r"pretend you are",
]

def sanitize_input(user_input: str) -> str:
    lower_input = user_input.lower()
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, lower_input):
            raise HTTPException(
                status_code=400,
                detail="잠재적으로 유해한 입력이 감지되었습니다."
            )
    # 최대 길이 제한
    if len(user_input) > 4000:
        raise HTTPException(status_code=400, detail="입력이 너무 깁니다.")
    return user_input.strip()

Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/api/chat")
@limiter.limit("10/minute")
async def chat(request: Request, chat_request: ChatRequest):
    # 분당 10회 제한
    pass

7. 멀티모달 입력 처리

이미지 분석 (GPT-4o Vision)

import base64
from pathlib import Path

async def analyze_image(image_path: str, question: str) -> str:
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")

    ext = Path(image_path).suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".png": "image/png", ".gif": "image/gif"}
    media_type = mime_map.get(ext, "image/jpeg")

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{media_type};base64,{image_data}"
                        },
                    },
                    {"type": "text", "text": question}
                ],
            }
        ],
    )
    return response.choices[0].message.content

Whisper API로 음성 처리

async def transcribe_audio(audio_file_path: str) -> str:
    with open(audio_file_path, "rb") as audio_file:
        transcript = await client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language="ko"
        )
    return transcript.text

8. 성능 최적화

Redis 응답 캐싱

import redis
import json
import hashlib

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

def get_cache_key(messages: list) -> str:
    content = json.dumps(messages, sort_keys=True)
    return hashlib.md5(content.encode()).hexdigest()

async def cached_chat(messages: list) -> str:
    cache_key = get_cache_key(messages)
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages
    )
    result = response.choices[0].message.content

    # 1시간 TTL로 캐싱
    redis_client.setex(cache_key, 3600, json.dumps(result))
    return result

연결 풀링

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    echo=False,
)

AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

9. Docker Compose 배포

docker-compose.yml

version: '3.8'
services:
  backend:
    build: ./backend
    ports:
      - '8000:8000'
    environment:
      - OPENAI_API_KEY=your_key
      - DATABASE_URL=postgresql+asyncpg://user:pass@db/aiapp
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    restart: unless-stopped

  frontend:
    build: ./frontend
    ports:
      - '3000:3000'
    environment:
      - NEXT_PUBLIC_API_URL=http://backend:8000
    depends_on:
      - backend
    restart: unless-stopped

  db:
    image: pgvector/pgvector:pg16
    environment:
      - POSTGRES_DB=aiapp
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:

backend/Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

frontend/Dockerfile

FROM node:20-alpine AS builder

WORKDIR /app
COPY package*.json ./
RUN npm ci

COPY . .
RUN npm run build

FROM node:20-alpine AS runner
WORKDIR /app
COPY --from=builder /app/.next/standalone ./
COPY --from=builder /app/.next/static ./.next/static

EXPOSE 3000
CMD ["node", "server.js"]

배포 명령

# 빌드 및 시작
docker-compose up --build -d

# 로그 확인
docker-compose logs -f backend

# 스케일 아웃 (백엔드 3개 인스턴스)
docker-compose up --scale backend=3 -d

# 중지
docker-compose down

10. 퀴즈: 핵심 개념 확인

Q1. SSE(Server-Sent Events)를 AI 스트리밍에 사용하는 이유는?

정답: LLM이 토큰을 생성하는 즉시 클라이언트로 전달하여 사용자가 응답을 기다리는 동안 텍스트가 실시간으로 보이도록 하기 위해서입니다.

설명: WebSocket보다 단순하며 HTTP를 통한 단방향 서버에서 클라이언트로의 스트리밍에 최적입니다. FastAPI의 StreamingResponse와 프론트엔드의 EventSource API가 이 패턴을 지원합니다.

Q2. RAG(Retrieval-Augmented Generation)의 핵심 이점은?

정답: LLM의 훈련 데이터 한계를 극복하고 최신 또는 도메인 특화 정보를 제공할 수 있습니다.

설명: 모델을 재훈련하지 않고도 외부 문서를 검색하여 컨텍스트에 포함시킵니다. 벡터 유사도 검색으로 관련 문서를 찾고, 이를 프롬프트에 주입하여 정확한 답변을 생성합니다. 환각(hallucination) 현상을 줄이는 효과도 있습니다.

Q3. FastAPI에서 Pydantic 모델을 사용하는 주요 이유는?

정답: 요청 및 응답 데이터의 자동 검증, 직렬화, API 문서 자동 생성을 위해서입니다.

설명: Pydantic은 Python 타입 힌트를 기반으로 런타임에 데이터를 검증합니다. FastAPI는 이를 활용해 OpenAPI(Swagger) 문서를 자동으로 생성하며, 잘못된 입력에 대해 명확한 오류 메시지를 반환합니다.

Q4. 벡터 데이터베이스에서 HNSW 인덱스를 사용하는 이유는?

정답: 고차원 벡터 공간에서 근사 최근접 이웃(ANN) 검색을 빠르게 수행하기 위해서입니다.

설명: 수백만 개의 벡터를 브루트 포스 방식으로 비교하면 너무 느립니다. HNSW(Hierarchical Navigable Small World)는 계층적 그래프 구조로 검색 속도를 크게 향상시키면서 높은 정확도를 유지합니다. pgvector, Chroma, Weaviate 등이 지원합니다.

Q5. LangChain의 ConversationBufferWindowMemory에서 k 파라미터의 역할은?

정답: 컨텍스트 창에 유지할 최근 대화 턴(turn)의 수를 지정합니다.

설명: LLM은 토큰 한계가 있으므로 모든 대화 기록을 보낼 수 없습니다. k=10이면 가장 최근 10번의 사용자-AI 교환을 유지하고 이전 내용은 삭제합니다. 메모리 비용과 컨텍스트 유지 간의 균형을 맞추는 중요한 파라미터입니다.

개요