Issue 26 | Frontend Integration: Architecture Planning for LangGraph's Frontend-Backend Separation

Updated on 4/17/2026

Welcome back, architects! I am your old friend.

Over the past 25 issues, we've been like a group of geeks cultivating in a basement. We have meticulously polished the "AI Content Agency", personally endowing the Planner with the ability to strategize, enabling the Researcher to scrape data across the web, allowing the Writer to produce brilliant prose, and giving the Editor a piercing eye for detail.

Watching the green text scrolling across the terminal, you might think: "Wow, this is so cool!"

But wake up, students! Your boss, your clients, and the end-users will absolutely never open a terminal to type commands! What they need is a beautiful webpage, an input box, a blinking loading animation, and finally, with a "ding", a perfect viral article appearing on the screen.

Starting from today's issue, we are bringing our AI Agency to the front stage. What we will explore is: how to dress LangGraph, an extremely time-consuming "slow-thinking" engine, in a standard, elegant Web API jacket, and allow the frontend (like React/Vue) to interact with it comfortably.

Don't think this is simple. If you dare to directly await graph.invoke() in FastAPI or Express, I guarantee your frontend will crash due to HTTP timeouts, and your users will stare at a white screen doubting their lives.

Are you ready? Today we will not only discuss the "theory", but also hand-code the "practice".

🎯 Learning Objectives for This Issue

  1. Cognitive Upgrade: Understand the fundamental conflict between the LangGraph state machine and the traditional HTTP request-response model.
  2. Architecture Design: Master the frontend-backend separation architecture of "Asynchronous Task Dispatching + State Polling (Heartbeat)".
  3. Backend Practice: Use FastAPI to build a standard LangGraph wrapper layer, perfectly binding the Task ID with LangGraph's Thread ID.
  4. Frontend Integration: Write the heartbeat polling logic on the React (TypeScript) side to achieve a silky-smooth user experience.

📖 Principle Analysis

In traditional Web development, typical CRUD interfaces are synchronous: Frontend sends a request -> Backend queries the database -> Returns the result. The entire process is usually completed within 200 milliseconds.

But what about our AI Content Agency? The Planner needs to brainstorm an outline (3 seconds), the Researcher needs to search Google and summarize (10 seconds), the Writer needs to write a 2000-word long article (20 seconds), and the Editor needs to review and revise (10 seconds). A complete Graph run might take 40 seconds to 1 minute!

If you use traditional synchronous HTTP requests: The frontend sends a POST request and then waits foolishly. The browser's default HTTP timeout is usually 30 to 60 seconds. If the network jitters slightly, or the LLM API acts up, the connection will drop. The user sees a 504 Gateway Timeout, while your backend is actually still working hard running the model, and the generated result ultimately cannot be returned to the frontend. This is a typical case of "money spent (Tokens consumed), work done, but the customer ran away".

Solution: Restaurant Pager Pattern (Asynchronous Polling)

We need to introduce an "asynchronous task + polling" mechanism. Just like ordering food at KFC:

  1. You order a "Family Bucket" at the front counter (submit the creation topic).
  2. The cashier won't make you stand at the counter waiting, but gives you an order receipt/pager (Task ID) and tells you: "Go sit over there and watch the screen." (Backend immediately returns 202 Accepted).
  3. You (the frontend) look up at the big screen every few seconds (send a GET request to query the status).
  4. When the big screen shows your order number, you take your receipt to pick up your food (fetch the final article).

Combined with LangGraph, we have a fantastic advantage: LangGraph natively supports Checkpointer (persistence), and its thread_id can naturally serve as our Task ID!

Let's look at the flow chart of this architecture:

sequenceDiagram
    participant U as User (React UI)
    participant API as FastAPI (Backend Gate)
    participant BG as Background Task / Worker
    participant LG as LangGraph (AI Agency)
    participant DB as Checkpointer (SQLite/Redis)

    U->>API: 1. POST /api/agency/generate {topic: "The Future of AI"}
    API->>DB: 2. Generate thread_id (Task ID)
    API->>BG: 3. Trigger background async execution of Graph
    API-->>U: 4. Return 202 Accepted {task_id: "1234-5678"}
    
    rect rgb(240, 248, 255)
    Note over U, API: Frontend Heartbeat Polling
    loop Execute every 3 seconds
        U->>API: 5. GET /api/agency/status/1234-5678
        API->>DB: 6. Read the latest State of this thread_id
        alt Task is running
            DB-->>API: State: {status: "Researcher collecting..."}
            API-->>U: Return 200 {status: "running", data: null}
        else Task is completed
            DB-->>API: State: {status: "done", final_article: "..."}
            API-->>U: Return 200 {status: "completed", article: "..."}
        end
    end
    end
    
    %% Background execution flow
    BG->>LG: Async call graph.ainvoke(..., config={"configurable": {"thread_id": "1234-5678"}})
    LG->>DB: Real-time update of State at each step

See that? FastAPI is only responsible for receiving guests and checking status. The real heavy lifting is handed over to background tasks, and LangGraph's Checkpointer becomes the bridge for state synchronization between the frontend and backend.

💻 Practical Code Drill

To let everyone get it running directly, I will provide two parts of the code: Python Backend (FastAPI + LangGraph) and TypeScript Frontend (React).

1. Backend: FastAPI + LangGraph Wrapper Layer

First, we need a simulated Agency Graph. To keep the code from being too bloated, I'll use a simplified StateGraph to represent our complex Planner/Researcher/Writer workflow, and introduce MemorySaver as the persistence layer.

Install dependencies:

pip install fastapi uvicorn langgraph langchain-openai pydantic

main.py core code:

import asyncio
import uuid
from typing import Dict, TypedDict, Any
from fastapi import FastAPI, BackgroundTasks, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# ==========================================
# 1. Define LangGraph State and Nodes (Simulating our AI Agency)
# ==========================================
class AgencyState(TypedDict):
    topic: str
    current_agent: str      # The Agent currently working
    draft: str              # Draft
    final_article: str      # Final article
    status: str             # Task status: "running", "completed", "failed"

# Simulate Planner Node
async def planner_node(state: AgencyState):
    print(f"[Planner] Brainstorming outline for topic '{state['topic']}'...")
    await asyncio.sleep(2) # Simulate LLM thinking time
    return {"current_agent": "Planner", "draft": "Outline: 1. Background 2. Development 3. Conclusion"}

# Simulate Writer Node
async def writer_node(state: AgencyState):
    print(f"[Writer] Writing first draft based on outline...")
    await asyncio.sleep(3) # Simulate LLM writing time
    return {"current_agent": "Writer", "draft": state["draft"] + "\nBody: AI is reshaping the world..."}

# Simulate Editor Node
async def editor_node(state: AgencyState):
    print(f"[Editor] Polishing the article...")
    await asyncio.sleep(2)
    final_text = state["draft"] + "\n[Proofreading complete, ready to publish]"
    return {"current_agent": "Editor", "final_article": final_text, "status": "completed"}

# Build workflow graph
workflow = StateGraph(AgencyState)
workflow.add_node("planner", planner_node)
workflow.add_node("writer", writer_node)
workflow.add_node("editor", editor_node)

workflow.add_edge(START, "planner")
workflow.add_edge("planner", "writer")
workflow.add_edge("writer", "editor")
workflow.add_edge("editor", END)

# Core: Introduce Checkpointer so every step's state in the graph can be saved and queried!
memory = MemorySaver()
agency_graph = workflow.compile(checkpointer=memory)

# ==========================================
# 2. FastAPI Wrapper Layer (Web API)
# ==========================================
app = FastAPI(title="AI Content Agency API")

# Allow frontend CORS requests
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Request Model
class GenerateRequest(BaseModel):
    topic: str

# Async background task: Responsible for actually executing LangGraph
async def run_agency_graph_background(thread_id: str, topic: str):
    config = {"configurable": {"thread_id": thread_id}}
    initial_state = {
        "topic": topic,
        "current_agent": "Initializing",
        "draft": "",
        "final_article": "",
        "status": "running"
    }
    try:
        # Use ainvoke for async execution, won't block FastAPI main thread
        # Note: Here we run it all at once. Because checkpointer is configured,
        # LangGraph will automatically write each step's state into memory in the background.
        await agency_graph.ainvoke(initial_state, config=config)
    except Exception as e:
        print(f"Graph execution failed: {e}")
        # If an error occurs, we need to manually update the status to failed (in actual production, a dedicated exception handling node is recommended)
        agency_graph.update_state(config, {"status": "failed"})

@app.post("/api/v1/agency/generate", status_code=202)
async def start_generation(req: GenerateRequest, background_tasks: BackgroundTasks):
    """
    Frontend calls this endpoint to submit a task and get a pager (task_id)
    """
    # 1. Generate a unique Task ID (equivalent to LangGraph's Thread ID)
    task_id = str(uuid.uuid4())
    
    # 2. Throw the time-consuming Graph execution into a background task
    background_tasks.add_task(run_agency_graph_background, task_id, req.topic)
    
    # 3. Immediately return Task ID to the frontend
    return {
        "message": "Task accepted. Please poll the status.",
        "task_id": task_id
    }

@app.get("/api/v1/agency/status/{task_id}")
async def get_status(task_id: str):
    """
    Frontend polls this endpoint via heartbeat to get the latest progress
    """
    config = {"configurable": {"thread_id": task_id}}
    
    # Read the latest State from Checkpointer
    state_snapshot = agency_graph.get_state(config)
    
    # If no snapshot is found, the task hasn't started or the ID is wrong
    if not state_snapshot or not state_snapshot.values:
        raise HTTPException(status_code=404, detail="Task not found or not started yet.")
    
    current_state = state_snapshot.values
    
    # Construct friendly data to return to the frontend
    response = {
        "task_id": task_id,
        "status": current_state.get("status", "running"),
        "current_agent": current_state.get("current_agent", "Unknown"),
    }
    
    # If completed, return the final article
    if current_state.get("status") == "completed":
        response["final_article"] = current_state.get("final_article")
        
    return response

if __name__ == "__main__":
    import uvicorn
    # Run: python main.py
    uvicorn.run(app, host="0.0.0.0", port=8000)

2. Frontend: Heartbeat Polling on the React Side

Now, the backend is ready. The frontend no longer needs to wait foolishly for 1 minute. We will use TypeScript + React hooks to write an elegant polling mechanism.

// Frontend: React Component (AgencyClient.tsx)
import React, { useState, useEffect, useRef } from 'react';

// Define interface types
interface TaskResponse {
  task_id: string;
  status: 'running' | 'completed' | 'failed';
  current_agent: string;
  final_article?: string;
}

export const AgencyClient: React.FC = () => {
  const [topic, setTopic] = useState<string>('');
  const [taskId, setTaskId] = useState<string | null>(null);
  const [status, setStatus] = useState<string>('idle');
  const [agentMsg, setAgentMsg] = useState<string>('');
  const [article, setArticle] = useState<string>('');
  
  // Use ref to store timer ID for easy cleanup
  const pollingIntervalRef = useRef<NodeJS.Timeout | null>(null);

  // 1. Submit task (Order food)
  const handleGenerate = async () => {
    setStatus('submitting');
    setArticle('');
    try {
      const res = await fetch('http://localhost:8000/api/v1/agency/generate', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ topic })
      });
      const data = await res.json();
      setTaskId(data.task_id); // Got the pager!
      setStatus('running');
    } catch (error) {
      console.error("Submission failed", error);
      setStatus('failed');
    }
  };

  // 2. Listen for taskId changes, start heartbeat polling
  useEffect(() => {
    if (!taskId || status !== 'running') return;

    // Define polling function
    const pollStatus = async () => {
      try {
        const res = await fetch(`http://localhost:8000/api/v1/agency/status/${taskId}`);
        if (!res.ok) return;
        
        const data: TaskResponse = await res.json();
        
        // Update UI to show which Agent is currently working
        setAgentMsg(`Current processing node: ${data.current_agent}...`);

        if (data.status === 'completed') {
          // Task completed, stop polling, show article
          setStatus('completed');
          setArticle(data.final_article || '');
          if (pollingIntervalRef.current) clearInterval(pollingIntervalRef.current);
        } else if (data.status === 'failed') {
          setStatus('failed');
          if (pollingIntervalRef.current) clearInterval(pollingIntervalRef.current);
        }
      } catch (error) {
        console.error("Polling failed", error);
      }
    };

    // Check immediately once, then check every 2 seconds (Heartbeat: 2000ms)
    pollStatus();
    pollingIntervalRef.current = setInterval(pollStatus, 2000);

    // Clean up timer on component unmount
    return () => {
      if (pollingIntervalRef.current) clearInterval(pollingIntervalRef.current);
    };
  }, [taskId, status]);

  return (
    <div style={{ padding: '20px', maxWidth: '600px', margin: '0 auto' }}>
      <h2>AI Content Agency Client</h2>
      
      <div style={{ display: 'flex', gap: '10px', marginBottom: '20px' }}>
        <input 
          type="text" 
          value={topic} 
          onChange={(e) => setTopic(e.target.value)} 
          placeholder="Enter article topic, e.g., The Future of AI"
          disabled={status === 'running'}
          style={{ flex: 1, padding: '8px' }}
        />
        <button 
          onClick={handleGenerate} 
          disabled={!topic || status === 'running'}
        >
          {status === 'running' ? 'Creating...' : 'Start Creation'}
        </button>
      </div>

      {/* Status Display Area */}
      {status === 'running' && (
        <div style={{ color: 'blue', fontStyle: 'italic' }}>
          <p>⏳ Grinding out the draft for you, please wait...</p>
          <p>🤖 {agentMsg}</p>
        </div>
      )}

      {/* Result Display Area */}
      {status === 'completed' && (
        <div style={{ border: '1px solid #ccc', padding: '15px', borderRadius: '8px', backgroundColor: '#f9f9f9' }}>
          <h3>🎉 Creation Complete:</h3>
          <pre style={{ whiteSpace: 'pre-wrap', fontFamily: 'inherit' }}>{article}</pre>
        </div>
      )}
    </div>
  );
};

Pitfalls and How to Avoid Them

Students, although the code above runs fine, there are a few major pitfalls in a production environment that you must know. As a mentor with 10 years of experience stepping into pitfalls, I need to give you a heads-up:

🚨 Pitfall 1: Memory Explosion (The Hidden Danger of MemorySaver)

In the Demo, we used MemorySaver() as the Checkpointer. This means the states of all tasks are saved in the memory of the FastAPI process. Consequence: If your website goes viral and has 10,000 tasks a day, your server memory will quickly burst. Once you restart the FastAPI service, all running task states will be completely lost, and the frontend will forever only poll a 404! How to Avoid: In a production environment, absolutely do not use MemorySaver. Please use AsyncSqliteSaver, AsyncPostgresSaver, or RedisSaver. Persist the state into a database, so you not only don't fear restarts but can also horizontally scale multiple FastAPI nodes.

🚨 Pitfall 2: The Blocking Crisis of BackgroundTasks

FastAPI's BackgroundTasks run in the same Event Loop by default. If a Node in your LangGraph contains synchronous blocking code (for example, using requests.get without async or CPU-intensive text processing), it will freeze the entire FastAPI application, causing the frontend's polling requests (GET status) to queue up and time out. How to Avoid:

  1. Ensure that every node in your Graph is truly asynchronous (use async def and asynchronous HTTP clients like httpx).
  2. If you must run synchronous, time-consuming legacy code, please use asyncio.to_thread() to execute it in a thread pool, or simply introduce Celery / Redis Queue (RQ) to completely separate the tasks into independent Worker processes.

🚨 Pitfall 3: DDoSing Yourself (Polling Frequency Too High)

If the frontend's setInterval is set to 100 milliseconds, and 100 users order food at the same time, your backend will have to endure 1,000 queries per second. How to Avoid: Introduce an Exponential Backoff strategy. Start by checking every 2 seconds; if it's not done after 10 seconds, change to checking every 5 seconds; after another 30 seconds, change to checking every 10 seconds. Don't foolishly keep requesting the server at a high frequency.

📝 Summary of This Issue

Today, we successfully led the AI Content Agency out of the terminal basement, put on the FastAPI suit, and completed the first "handshake" with the React frontend.

We learned:

  1. The long-running nature of LangGraph dictates that we must adopt an asynchronous flow with frontend-backend separation.
  2. Restaurant Pager Pattern: POST triggers the task and returns a Task ID, GET polls the status.
  3. LangGraph's Killer Feature: Directly utilizing checkpointer and thread_id perfectly achieves the binding of the business Task ID with the underlying state graph. The backend doesn't need to maintain an extra state dictionary by itself at all!

Teaser for Next Issue: Although polling is useful, it's not "sexy" enough. Have you noticed that when ChatGPT generates content, it pops out word by word (typewriter effect)? In Issue 27, we will take on an advanced challenge: abandon polling, introduce SSE (Server-Sent Events) and Streaming output, allowing your frontend to see every single word typed by the Writer Agent in real-time!

Architects, get today's code running, and see you next time! Don't forget to add Postgres persistence to your APIs!