FastAPI 기반 AI 챗봇 백엔드 로직 설계: 상태 관리와 스트리밍

 FastAPI 기반 AI 챗봇 백엔드 로직 설계: 상태 관리와 실시간 스트리밍(SSE) 아키텍처
최근 AI 기반의 백엔드 시스템을 직접 설계하고 구현하면서 뼈저리게 느낀 점이 하나 있습니다. "OpenAI API를 단순히 한 번 호출하는 것은 주니어 개발자도 할 수 있지만, 진짜 프로덕션 레벨의 서비스 퀄리티를 만드는 것은 결국 백엔드 엔지니어의 아키텍처 설계 역량"이라는 사실입니다.
ChatGPT처럼 이전 대화의 문맥을 완벽하게 기억하고, 마치 사람이 타이핑하듯 실시간으로 응답하는 챗봇을 만들려면 단순한 1회성 API 호출 로직으로는 턱없이 부족합니다.
오늘은 HTTP 프로토콜의 본질적인 한계인 '무상태성(Stateless)'을 극복하고, 유저에게 끊김 없는 매끄러운 사용자 경험(UX)을 제공하기 위한 AI 백엔드 아키텍처를 FastAPI 스택으로 설계해 보겠습니다.

 

1. HTTP 무상태성(Stateless)의 한계와 컨텍스트(Context) 관리의 필요성
웹 서버가 클라이언트와 통신하는 HTTP 프로토콜은 본질적으로 '기억력'이 없습니다. 클라이언트가 "안녕"이라고 인사하고, 이어서 "내 이름이 뭐게?"라고 물어도 서버는 직전에 누가 어떤 말을 했는지 전혀 기억하지 못합니다.
따라서 AI가 사람처럼 문맥을 이해하고 자연스럽게 대화하게 하려면, 백엔드 서버에서 **'대화 내역(History)'**을 안전하게 저장해 두어야 합니다. 그리고 새로운 질문이 들어올 때마다 과거의 대화 기록 전체를 배열(List) 형태로 묶어 OpenAI API에 함께 던져주어야 하죠. 우리는 이것을 전문 용어로 '컨텍스트 윈도우(Context Window)' 관리라고 부릅니다.

 

2. 대화 세션 유지를 위한 데이터베이스(DB) 설계 전략
그렇다면 이 방대한 대화 기록을 도대체 어디에 저장해야 할까요? 토이 프로젝트 수준이거나 사용자 수가 극히 적을 때는 파이썬의 메모리(딕셔너리 등)에 임시로 저장해도 동작은 합니다. 하지만 실제 서비스에서 서버가 재부팅되거나 로드밸런서로 서버가 여러 대 분산되는 순간, 유저의 대화 데이터는 모두 증발해 버립니다.
실무 백엔드 아키텍처에서는 목적에 따라 DB를 이원화하여 설계합니다.
 * RDBMS (PostgreSQL / MySQL): 대화 내역을 영구적으로 보존하고, 추후 유저의 행동 패턴이나 데이터를 분석해야 할 때 사용합니다. 하지만 매번 대화를 주고받을 때마다 디스크 기반의 DB에 접근하는 것은 서버에 엄청난 I/O 부하를 일으킵니다.
 * In-Memory DB (Redis): 현업에서 가장 많이 쓰이는 핵심 아키텍처입니다. 메모리 기반이라 매우 빠른 읽기/쓰기가 가능하므로, **현재 진행 중인 활성 세션(Active Session)**의 대화 기록을 캐싱해 두는 용도로 사용합니다. 일정 시간(TTL)이 지나면 자동으로 데이터를 지워주기 때문에 메모리 관리에도 탁월합니다.

 

3. 실무 방어 전략: SSE(Server-Sent Events)를 활용한 실시간 스트리밍
아키텍처 설계가 끝났다면, 이제 사용자 경험(UX)을 끌어올릴 차례입니다. LLM 모델이 전체 답변을 완성할 때까지 5~10초를 빈 화면으로 기다리게 만들면, 참을성 없는 유저들은 곧바로 서비스를 이탈해 버립니다.
ChatGPT처럼 한 글자씩 실시간으로 화면에 렌더링하려면 클라이언트와 서버 간의 지속적인 연결이 필요한데, 이때 FastAPI의 StreamingResponse와 OpenAI API의 stream=True 옵션을 결합하면 완벽한 스트리밍 서버를 구축할 수 있습니다.

 

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio

app = FastAPI()
client = AsyncOpenAI() # 비동기 클라이언트 사용 권장

# 스트리밍 데이터를 생성하는 제너레이터 함수
async def generate_chat_stream(prompt: str):
    stream_response = await client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True  # 🔥 핵심: 스트리밍 옵션 활성화
    )
    
    # 청크(Chunk) 단위로 쪼개진 데이터가 들어올 때마다 즉시 방출
    async for chunk in stream_response:
        if chunk.choices[0].delta.content is not None:
            # return이 아닌 yield를 사용하여 연결을 유지한 채 데이터 전송
            yield chunk.choices[0].delta.content
            await asyncio.sleep(0.01) # 버퍼링 과부하 조절

@app.get("/chat/stream")
async def chat_endpoint(message: str):
    # Server-Sent Events(SSE) 포맷으로 텍스트 스트리밍 반환
    return StreamingResponse(
        generate_chat_stream(message), 
        media_type="text/event-stream"
    )

코드 인사이트:
여기서 가장 눈여겨보아야 할 부분은 일반적인 return 대신 **yield**를 사용했다는 점입니다. 제너레이터(Generator) 패턴을 활용하여, OpenAI 서버에서 단어(Chunk)가 하나씩 도착할 때마다 백엔드 서버가 이를 쥐고 있지 않고 클라이언트(프론트엔드)로 즉시 토스해 버리는 역할을 합니다. media_type="text/event-stream"을 명시해 주어야 프론트엔드에서 SSE로 정상 인식합니다.

 

실무 대처법: 스트리밍 중 클라이언트 연결 끊김(Disconnect) 에러 처리
FastAPI로 스트리밍 서버를 띄워놓고 실제 유저 트래픽을 받아보면 백이면 백 발생하는 에러가 있습니다. AI가 한참 답변을 생성해서 yield로 쏴주고 있는데, 유저가 답변이 느리다며 '뒤로 가기'를 누르거나 브라우저 창을 닫아버리는 경우입니다.
이때 백엔드에서는 갈 곳 잃은 데이터 때문에 asyncio.CancelledError가 터지면서 서버 리소스가 질질 새는(Memory Leak) 현상이 발생합니다.
이를 방지하려면 제너레이터 함수(generate_chat_stream) 내부에 try-except 블록을 걸어두고, CancelledError가 감지되면 즉시 OpenAI API 호출 세션을 강제로 종료(break 또는 자원 해제)시키는 방어 로직을 반드시 추가해야 합니다. 이 작은 예외 처리 하나가 서버가 뻗는 것을 막아줍니다.

 

4. 마무리 (백엔드 엔지니어의 시야)
단순해 보이는 챗봇 화면 뒤에는 이토록 치열한 백엔드의 고민이 숨어 있습니다.
위에서 다룬 실시간 스트리밍(SSE) 로직과 Redis 기반의 빠르고 안전한 컨텍스트(Context) 세션 관리가 결합되면, 대용량 트래픽이 몰려와도 끄떡없는 프로덕션 레벨의 AI 챗봇 백엔드가 완성됩니다.
하지만 서비스를 오픈하고 나면 또 다른 거대한 벽을 마주하게 됩니다. 바로 전 세계 해커들의 '프롬프트 인젝션' 공격이죠. 다음 포스팅에서는 이렇게 공들여 만든 AI 시스템을 악의적인 공격으로부터 안전하게 지켜내는 다층 보안 전략 및 아키텍처에 대해 깊이 있게 다루어 보겠습니다.