Pillar 03 — LLM & Tech Stack · Cluster 3.3 · Engineers & CTOs · Technical

← Pillar 03: LLM & Tech Stack

Pillar 03 · Cluster 3.3 · LLM & Tech Stack Malaysia

FastAPI + LangGraph: Architecture untuk Production AI Agent Backend

LangGraph handles the agent logic. FastAPI handles everything else — routing, authentication, streaming, background tasks, and the API surface that the rest of your application talks to. Ini adalah production architecture, bukan tutorial hello world.

Consideration Technical 12 min read

.// TL;DR — Kalau busy, baca ni je dulu

FastAPI sebagai serving layer atas LangGraph adalah the pattern that separates production systems from notebook experiments. LangGraph manages agent state and execution; FastAPI manages HTTP concerns, auth, and streaming to clients.
Streaming is non-negotiable for agent UX. Users yang tunggu 30 saat untuk full response akan abandon. Server-Sent Events (SSE) via FastAPI lets you stream LangGraph output token-by-token as the agent works.
PostgreSQL sebagai checkpoint backend untuk LangGraph persistent state adalah the right default — not in-memory, not SQLite. Persistent state survives restarts and enables resume-after-failure.
Celery + Redis untuk background tasks yang long-running seperti tender document processing atau multi-step research workflows. Don't block the HTTP thread.
Realistic VPS sizing untuk single-tenant Malaysian SME deployment: 8 vCPU, 16GB RAM handles concurrent agent sessions without exotic infrastructure.

Most LangGraph tutorials show you how to define a graph and run it in a Jupyter notebook. That's useful for learning. But there's a meaningful gap between "it runs in my notebook" and "it runs reliably at 2am when 12 users are triggering concurrent agent sessions."

This article covers that gap. Specifically, how we structure the serving layer for LangGraph-based agents in production — using FastAPI as the HTTP interface, PostgreSQL for state persistence, and Celery for long-running background execution.

This is the architecture IRIS runs on. Bukan aspirational architecture — ini adalah benda yang sebenarnya deployed.

Why FastAPI specifically

Python's async framework landscape has several options — Django, Flask, FastAPI, Litestar. For AI agent backends, FastAPI wins on three specific dimensions:

Native async support. LangGraph agent execution is inherently async — waiting on LLM API calls, tool calls, database queries. FastAPI's ASGI foundation handles this cleanly. Flask's sync-by-default creates threading complexity when wrapping async agent code.
Streaming response support. FastAPI has first-class StreamingResponse and SSE support. Streaming LangGraph output through to the client requires this — it's not an afterthought bolted on.
Automatic schema generation. FastAPI generates OpenAPI docs from type hints. For an API that other services or frontend clients will consume, this means your API is self-documenting without extra work.

System architecture overview

Streaming agent output with Server-Sent Events

This is the most important piece to get right for user experience. An agent that processes a complex task might take 15–45 seconds. Without streaming, the user sees a blank screen and a spinner. With SSE streaming, they see the agent's reasoning and output appearing progressively — which transforms the UX from "broken?" to "working."

.// FastAPI SSE endpoint for streaming agent output

from fastapi import APIRouter, Depends from fastapi.responses import StreamingResponse from langgraph.graph import StateGraph import json router = APIRouter(prefix="/stream") @router.post("/agent/{thread_id}") async def stream_agent( thread_id: str, payload: AgentRequest, current_user: User = Depends(get_current_user), graph: CompiledGraph = Depends(get_graph), ): async def event_generator(): config = {"configurable": {"thread_id": thread_id}} # Stream events from LangGraph execution async for event in graph.astream_events( payload.dict(), config=config, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"].content if chunk: # SSE format: data: {json}\n\n yield f"data: {json.dumps({'type': 'token', 'content': chunk})}\n\n" elif event["event"] == "on_tool_start": yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name']})}\n\n" elif event["event"] == "on_chain_end": yield f"data: {json.dumps({'type': 'done'})}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, )

The X-Accel-Buffering: no header is important if you're behind Nginx as a reverse proxy. Without it, Nginx will buffer the SSE response and the client won't receive events until the buffer fills — which defeats the purpose of streaming entirely. This is a production gotcha that's easy to miss in local development where there's no proxy.

PostgreSQL checkpoint backend for persistent state

By default, LangGraph uses in-memory checkpointing. This means agent state is lost when the process restarts. For production, you need a persistent backend.

We use langgraph-checkpoint-postgres — the official Postgres checkpoint implementation. Setup is straightforward:

.// LangGraph with PostgreSQL checkpoint backend

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver from psycopg_pool import AsyncConnectionPool # Initialise connection pool at application startup async def lifespan(app: FastAPI): pool = AsyncConnectionPool( conninfo=settings.DATABASE_URL, max_size=20, kwargs={"autocommit": True, "prepare_threshold": 0}, ) checkpointer = AsyncPostgresSaver(pool) await checkpointer.setup() # Creates checkpoint tables if not exists # Compile graph with persistent checkpointer app.state.graph = workflow.compile(checkpointer=checkpointer) app.state.pool = pool yield # Application runs here await pool.close() app = FastAPI(lifespan=lifespan)

With PostgreSQL checkpointing, every agent state transition is persisted. This enables:

Resume after failure — if a worker crashes mid-execution, the agent can resume from last checkpoint rather than restart from scratch
Human-in-the-loop interrupts — agent pauses at a defined interrupt point, state is persisted, human reviews and approves, execution resumes from exactly where it paused
Full audit trail — every state checkpoint is queryable; you can replay the entire decision history for any agent thread
Multi-session continuation — user can close browser, come back tomorrow, and the thread continues from where they left off

Background tasks with Celery for long-running execution

Some agent tasks take minutes, not seconds — processing a 200-page tender document, running a full market intelligence sweep across multiple data sources, or coordinating a multi-step research workflow. Blocking an HTTP request thread for minutes is not acceptable.

The pattern: HTTP endpoint accepts the task, dispatches to Celery, returns a task ID immediately. Client polls /tasks/{task_id}/status or receives updates via SSE.

.// Submit long-running agent task to Celery

# FastAPI endpoint — returns immediately with task_id @router.post("/tasks/agent/run") async def submit_agent_task( payload: AgentTaskRequest, current_user: User = Depends(get_current_user), ): task = run_agent_task.delay( thread_id=payload.thread_id, input_data=payload.dict(), user_id=current_user.id, ) return {"task_id": task.id, "status": "queued"} # Celery task — runs in worker process @celery_app.task(bind=True, max_retries=3) def run_agent_task(self, thread_id: str, input_data: dict, user_id: str): import asyncio try: result = asyncio.run(execute_agent_graph(thread_id, input_data)) return {"status": "completed", "result": result} except Exception as exc: raise self.retry(exc=exc, countdown=30)

Authentication and multi-tenancy

For a SaaS AI agent system, every thread_id must be scoped to the authenticated subscriber. A subscriber must not be able to access or influence another subscriber's agent threads.

.// JWT auth middleware with subscriber scoping

from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") async def get_current_user(token: str = Depends(oauth2_scheme), db = Depends(get_db)): try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) subscriber_id = payload.get("sub") if not subscriber_id: raise HTTPException(status_code=401) except JWTError: raise HTTPException(status_code=401) return await db.get_subscriber(subscriber_id) # Thread ID must be prefixed with subscriber_id to prevent cross-tenant access def scoped_thread_id(subscriber_id: str, thread_id: str) -> str: return f"{subscriber_id}:{thread_id}"

Deployment on realistic Malaysian SME infrastructure

Not everyone has AWS enterprise contracts or Google Cloud credits. For early-stage Malaysian SaaS products, here's what actually works:

Component	Recommended setup	Monthly cost (est.)
Primary VPS	8 vCPU / 16GB RAM — Hetzner CX52 or DigitalOcean 8vCPU	RM 100–200
PostgreSQL	Same VPS (early stage) or managed DB (scale stage)	RM 0–200
Redis	Same VPS via Docker, or Upstash Redis serverless	RM 0–50
Object storage	Cloudflare R2 (no egress fees — important for Malaysia)	RM 0–20
Reverse proxy	Nginx on same VPS, Cloudflare in front for DDoS & CDN	RM 0
Process manager	supervisord managing Uvicorn workers + Celery workers	RM 0

Total infrastructure cost for early-stage: RM 100–470/month. This setup handles 20–30 concurrent agent sessions comfortably. Scale trigger: when you need more than that, split into dedicated database server first, then consider managed Kubernetes if complexity justifies it.

Why Cloudflare R2 for object storage: Standard AWS S3 charges for egress bandwidth — data transferred out to your users. In Malaysia, this can add up significantly for document-heavy agent workflows. R2 charges zero egress fees. For an agent system that regularly processes and serves PDF documents, this is a real cost difference.

What we use and why

.// TTD's production stack declaration

FastAPI (>=0.111) + LangGraph (>=0.2) + PostgreSQL 16 + Redis 7 + Celery 5. This is the exact stack IRIS runs on. Version pinning matters — LangGraph's API surface changed significantly between 0.1 and 0.2; don't upgrade without running your full test suite.

Uvicorn behind Nginx, managed by supervisord. Not Gunicorn with Uvicorn workers for this use case — the async nature of agent execution means a single Uvicorn process with proper concurrency handles our load better than the Gunicorn process model. Supervisord handles restarts and log management cleanly without Docker overhead on single-VPS deployments.

We do not use Docker in production for single-VPS deployments. Container overhead adds complexity and memory pressure that isn't worth it at this scale. Direct process management with supervisord is simpler to debug, easier to monitor, and boots faster after VPS restarts. Docker becomes worthwhile when you're orchestrating across multiple servers.

Alembic for database migrations. Every schema change — including LangGraph checkpoint table changes — goes through Alembic migrations. This is non-negotiable for production systems where you can't afford to recreate the database.

Common production failures and how to avoid them

Missing X-Accel-Buffering: no header. SSE streaming works locally, breaks behind Nginx. Add the header. Set proxy_buffering off in your Nginx location block for SSE routes.

LangGraph thread_id collision between tenants. If two subscribers happen to use the same thread_id, their state will mix. Always namespace thread IDs with subscriber ID. f"{subscriber_id}:{user_thread_id}" is the minimum safe pattern.

Celery task timeout with long-running agents. Default Celery task timeout is 30 seconds. Agent workflows that process large documents can run longer. Set task_soft_time_limit and task_time_limit explicitly based on your expected max execution time, and handle SoftTimeLimitExceeded gracefully.

PostgreSQL connection pool exhaustion. Under concurrent load, if each request opens its own connection rather than using the pool, Postgres will hit max_connections. Use psycopg_pool.AsyncConnectionPool with a max_size that's appropriate for your Postgres instance's max_connections setting.

LangGraph version upgrades breaking checkpoints. The checkpoint schema can change between LangGraph versions. Test upgrades on a copy of production data before applying to production. Checkpoint table incompatibility is silent — the agent may start from scratch rather than resume, without throwing an obvious error.

Takeaway

The gap between a working LangGraph notebook and a production AI agent backend is real but bridgeable with the right architecture decisions. FastAPI handles the HTTP layer correctly for async streaming workloads. PostgreSQL gives you the persistent state that production reliability requires. Celery decouples long-running execution from HTTP request cycles.

Start with the simplest version: FastAPI + LangGraph + PostgreSQL on a single VPS. Get it working. Monitor it. Add Celery when you have long-running tasks that need it. Add horizontal scaling when your metrics show you need it. Avoid over-engineering for a scale you don't have yet.

Nak build production AI agent backend untuk business korang?

We design and build the full stack — LangGraph agent architecture, FastAPI serving layer, deployment on infrastructure yang realistic untuk scale dan budget korang. Book a session.

Book Teh Tarik Session ❯❯

WRITTEN FOR

Backend engineers building AI systems
CTOs evaluating production architecture
Tech leads moving from notebook to prod
Anyone deploying LangGraph at scale

NOTIFY ME

Get notified when new articles ship.