XGen Backend Architecture Design — A FastAPI-Based AI Platform
This post summarizes the experience of designing the backend for the XGen platform (formerly PlateeRAG) at Plateer AI-LAB. I'll walk through the process of building the architecture around FastAPI to meet the specific requirements of an AI workflow platform.
Why FastAPI
The two most critical requirements for an AI platform backend were async processing and streaming responses. LLM calls take anywhere from a few seconds to tens of seconds, so a synchronous approach was never on the table.
from fastapi import FastAPI, APIRouter
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# 앱 시작 시 DB 연결, Redis 초기화
await init_database()
await init_redis_pool()
yield
# 앱 종료 시 리소스 정리
await close_database()
await close_redis_pool()
app = FastAPI(
title="XGen Platform API",
version="1.0.0",
lifespan=lifespan,
)
Overall Architecture
The project structure is cleanly separated by responsibility.
backend/
├── api/
│ ├── routes/ # 라우터 정의
│ ├── dependencies/ # 의존성 주입
│ └── middlewares/ # 미들웨어
├── core/
│ ├── config.py # 환경설정
│ ├── security.py # 인증/인가
│ └── events.py # 이벤트 핸들러
├── models/ # SQLAlchemy 모델
├── schemas/ # Pydantic 스키마
├── services/ # 비즈니스 로직
│ ├── workflow/ # 워크플로우 실행
│ ├── nodes/ # 노드 처리기
│ └── storage/ # 파일 저장소
└── utils/ # 유틸리티
Core Design Principles
1. Dependency Injection
FastAPI's Depends is used extensively to inject the service layer.
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
async def get_workflow_service(
db: AsyncSession = Depends(get_db),
) -> WorkflowService:
return WorkflowService(db)
2. Unified Error Handling
A custom exception handler ensures all API responses follow a consistent shape.
class AppException(Exception):
def __init__(self, status_code: int, detail: str, code: str):
self.status_code = status_code
self.detail = detail
self.code = code
@app.exception_handler(AppException)
async def app_exception_handler(request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.code, "message": exc.detail},
)
Lessons Learned
After 303 commits, the importance of getting the initial design right was driven home once again. The workflow execution engine in particular started out with a simple design, but when parallel execution requirements were added, it needed a major refactor. FastAPI's async/await model made asynchronous processing straightforward, but managing complex workflow state remained a genuine challenge.
The next post will cover the backend implementation of the canvas node editor — specifically the topological-sort-based workflow executor.