LLM 앱 개발을 시작할 때 갖고 싶었던 FastAPI 백엔드 보일러플레이트
요약
LLM 앱 개발 시 반복적으로 발생하는 FastAPI 백엔드 문제를 해결한 완전한 스캐폴드를 공개했습니다. 이 스캐폴드는 스트리밍, 장시간 에이전트 실행 처리, WebSocket 세션 관리 등 핵심 기능을 포함하여 개발 시간을 획기적으로 단축시킵니다.
핵심 포인트
- FastAPI 기반의 AI Agent 백엔드 스캐폴드가 제공됩니다.
- 스트리밍 응답을 위한 비동기 제너레이터와 StreamingResponse를 사용합니다.
- 장시간 에이전트 실행은 폴링(polling) 패턴으로 처리하여 타임아웃 문제를 해결합니다.
- WebSocket 세션 상태 관리를 용이하게 하는 구조가 포함되어 있습니다.
제가 구축했던 모든 LLM 백엔드는 같은 방식으로 시작했습니다.
FastAPI를 엔드포인트로 사용하고, 기본적인 /chat 라우트를 만들고, 그리고 이전 프로젝트에서 이미 해결했던 문제들을 다시 푸는 데 일주일이 걸렸습니다. 이벤트 루프를 막지 않으면서 응답을 스트리밍하는 방법은 무엇일까요? 에이전트 실행 시간이 90초가 걸릴 때 리버스 프록시가 이를 종료시키면 어떻게 될까요? WebSocket 세션의 대화 기록을 사용자 간에 누수 없이 유지하려면 어떻게 해야 할까요?
이것들은 어려운 문제가 아닙니다. 단지 지루할 뿐입니다. 한 번 해결하면 프로젝트마다 복사 붙여넣기만 하면 되는데, 이전 솔루션이 미완성이라서 적응하는 데 시간을 보내게 되는 종류의 문제입니다.
저는 이 전체 솔루션을 FastAPI + AI Agent 백엔드 스캐폴드로 패키징했습니다.
개요
AI 에이전트 백엔드를 구축하는 팀을 위한 완전한 FastAPI 애플리케이션 스캐폴드입니다. 튜토리얼이 아닙니다. 테스트, Docker 설정, 그리고 모든 설정 옵션에 대한 문서화가 포함된 작동 코드입니다.
app/
main.py — FastAPI 팩토리, 라이프스팬(lifespan), 미들웨어 스택
config.py — Pydantic BaseSettings (환경 변수, .env 파일)
...
이 스캐폴드가 해결하는 네 가지 문제
1. 블로킹하지 않는 스트리밍
순진한 접근 방식은 여러 클라이언트로 동시에 스트리밍을 시도할 때까지 작동합니다:
# 순진한 방식: 생성 중에 블록함
@app.post("/chat")
async def chat(request: ChatRequest):
...
스캐폴드의 스트리밍 엔드포인트는 적절한 비동기 제너레이터와 StreamingResponse를 사용합니다. 이는 블로킹이 없고, 백프레셔(backpressure)가 처리되며, nginx용으로 Cache-Control 및 X-Accel-Buffering 헤더가 올바르게 설정됩니다:
@router.post("/completions")
async def chat_completions(request: ChatRequest, llm: LLMService = Depends(get_llm)):
if request.stream:
...
SSE를 읽는 모든 프런트엔드(Vercel AI SDK, 브라우저의 EventSource, Python의 httpx)와 작동합니다.
2. 타임아웃되지 않는 장시간 에이전트 실행
2. 타임아웃되지 않는 장시간 에이전트 실행
대부분의 리버스 프록시(reverse proxy) 기본 설정은 90초 동안의 에이전트 실행을 중단시킵니다. 일반적인 해결책(타임아웃 시간 전역 증가)은 새로운 문제를 야기합니다. 즉, 느린 에이전트 실행과 정지된 프로세스를 더 이상 구별할 수 없게 됩니다.
스캐폴드(scaffold)의 백그라운드 작업 패턴은 task_id와 함께 즉시 반환됩니다. 클라이언트는 이 결과를 폴링(poll)합니다:
# POST /agents/run — 100ms 미만으로 반환
@router.post("/run", status_code=202)
async def run_agent(request: AgentRunRequest, req: Request, llm: LLMService = Depends(get_llm)):
...
외부 브로커가 필요 없습니다. TaskQueue는 asyncio와 세마포어 기반 동시성 제한(MAX_BACKGROUND_TASKS 환경 변수, 기본값 10)을 사용합니다. 분산 워커(distributed workers)가 필요한 경우 Celery나 ARQ로 교체할 수 있지만 인터페이스는 동일하게 유지됩니다.
3. WebSocket 다중 턴 세션
WebSocket 에이전트 세션의 까다로운 부분은 연결 자체가 아니라 세션 상태입니다. 대화 기록은 재연결(reconnects)을 거쳐도 지속되어야 하며, session_id별로 격리되어야 하고, 세션이 유휴 상태가 되면 정리되어야 합니다.
@router.websocket("/chat")
async def websocket_chat(
websocket: WebSocket,
...
인메모리(in-memory) 세션 저장소(_sessions: dict[str, list[ChatMessage]])는 단일 인스턴스 배포에 적합합니다. 다중 인스턴스의 경우 Redis 기반 저장소로 교체하면 됩니다. 인터페이스는 변경되지 않습니다.
4. 재시작 후에도 유지되는 사용자별 속도 제한(Rate Limiting)
재시작 시 초기화되는 고정 비율 제한기는 프로덕션 환경에서 유용하지 않습니다. 이 스캐폴드는 Redis 기반 토큰 버킷을 사용합니다:
# 60 requests/minute, burst of 10 — RATE_LIMIT_RPM 및 RATE_LIMIT_BURST를 통해 구성 가능
# Redis를 사용할 수 없는 경우 인메모리로 폴백(fallback) (단일 인스턴스 안전)
class RateLimitMiddleware(BaseHTTPMiddleware):
...
속도 제한 키는 Authorization 헤더에서 파생됩니다(해시 처리되므로 원본 토큰이 Redis에 나타나지 않습니다). 인증되지 않은 요청의 경우 클라이언트 IP로 폴백합니다.
실제로 쿼리 가능한 구조화된 로깅(Structured logging)
실제로 쿼리 가능한 구조화된 로깅(Structured logging)
요청마다 request_id가 부여됩니다. 응답은 status_code와 latency_ms를 기록합니다. 모든 출력은 기본적으로 JSON 형식입니다:
{"timestamp": "2026-09-15T14:00:01", "level": "info", "request_id": "a1b2c3d4", "method": "POST", "path": "/chat/completions"}
{"timestamp": "2026-09-15T14:00:02", "level": "info", "request_id": "a1b2c3d4", "status_code": 200, "latency_ms": 1840}
커스텀 파서 없이 Datadog, CloudWatch 또는 Grafana Loki에 직접 가져올 수 있습니다.
로컬 개발 환경에서는 사람이 읽기 쉬운 형식으로 전환하려면 LOG_JSON=false를 사용하세요.
에이전트 로직 추가하기
스캐폴드는 인프라를 처리합니다. 사용자 에이전트 로직은 services/llm.py에 작성하면 됩니다:
class LLMService:
async def chat(self, messages: list[ChatMessage], model=None, system=None, **kwargs) -> str:
# 여기에 시스템 프롬프트, 도구, 다단계 로직을 추가하세요
...
도구를 추가하고, 호출 체인(chain calls)을 만들고, 재시도 로직(retry logic)을 구현해도 엔드포인트는 그대로 유지됩니다. 에이전트의 복잡성은 한 곳에 존재합니다.
테스트 설정
제공된 테스트 피처(test fixtures)를 사용하면 실제 LLM API나 Redis 없이 모든 엔드포인트를 테스트할 수 있습니다:
def test_non_streaming_chat(client, mock_llm):
mock_llm.queue_response("Hello! How can I help?")
...
MockLLMService는 큐에 저장된 응답을 순서대로 반환합니다. 내부 메서드를 모킹(mocking)하거나 OpenAI SDK를 원숭이 패치(monkeypatching)할 필요가 없습니다.
빠른 시작
cp .env.example .env
# .env에 OPENAI_API_KEY 또는 ANTHROPIC_API_KEY 설정
...
이 스캐폴드는 OpenAI와 Anthropic을 기본적으로 지원합니다. .env 파일에서 LLM_PROVIDER=anthropic으로 변경하여 제공업체를 전환할 수 있습니다.
스캐폴드 받기
**FastAPI + AI Agent Backend Scaffold**는 Gumroad에서 $69에 구매 가능합니다 — 일회성 구매, GitHub 레포지토리 제공, MIT 라이선스입니다.
전체 애플리케이션(app/, tests/, docker/), requirements.txt, pyproject.toml, 그리고 .env.example이 포함되어 있습니다.
만약 이와 함께 에이전트 테스트를 구축한다면, Pytest for AI Agents Starter Kit은 동일한 MockLLMService 패턴을 사용합니다 — 피처가 이 스캐폴드(scaffold)의 테스트 설정과 직접 구성됩니다.
AI 자동 생성 콘텐츠
본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.
원문 바로가기