Most LangGraph tutorials end when the demo works in a Jupyter notebook. That’s the wrong place to stop. The gap between a working prototype and a production AI agent that handles concurrent users, survives process restarts, and returns results in time is where real engineering happens — and where most teams quietly fail.
This guide covers the full stack: LangGraph for deterministic workflow orchestration, Redis for distributed state and checkpoint persistence, and FastAPI for async request handling. I’ll cover the architectural decisions that actually matter, the failure modes that will bite you in production, and the patterns that have proven durable under load.
This is not a beginner walkthrough. If you want to understand why each layer is here and what breaks when you remove it, read on.
Why These Three Technologies, Together
Before diving into implementation, it’s worth being precise about what problem each layer solves — because using them without that clarity leads to architectures that look right but fail under production conditions.
LangGraph is a stateful graph execution engine built on top of LangChain. Its value is not just the graph abstraction — it’s the checkpoint protocol. Every node execution can be checkpointed, which means execution can be paused, inspected, resumed, or replayed. For agents that run long-horizon tasks (research pipelines, code generation workflows, multi-step reasoning chains), this is what separates a recoverable system from an expensive pile of lost compute.
Redis in this stack wears two hats. As a checkpoint backend, it stores LangGraph’s state snapshots durably and fast — avoiding the write amplification you’d get with a relational database when checkpointing at high frequency. As a pub/sub and task queue substrate, it enables async agent execution patterns that decouple request acceptance from agent execution time.
FastAPI handles the API surface. Its async-first design maps cleanly onto LLM workloads, where nearly all latency is I/O-bound (network calls to model providers, tool API calls, database lookups). More practically: FastAPI’s native support for Server-Sent Events (SSE) and WebSockets gives you clean streaming primitives without bolting on additional infrastructure.
Together, these three form a coherent harness: FastAPI accepts requests and streams responses, LangGraph executes agent logic with checkpointed state, and Redis holds that state across process boundaries. Each layer has a single, clear responsibility. That clarity is what makes the system debuggable when things go wrong at 2am.
Architectural Overview
The production agent architecture follows a request-execution-response cycle with async decoupling at the boundary:
Client Request
│
▼
FastAPI (async endpoint)
│
├── Validate input
├── Create thread_id
├── Enqueue agent task (Redis queue)
└── Return task_id + SSE stream URL
│
▼
Worker Process
│
├── Dequeue task
├── Load checkpoint from Redis (if resuming)
├── Execute LangGraph nodes
│ ├── Checkpoint each node to Redis
│ └── Publish progress events (Redis pub/sub)
└── Write final result to Redis
│
▼
FastAPI SSE Handler
│
├── Subscribe to Redis channel for thread_id
└── Stream events to client
This architecture makes a deliberate trade-off: it adds latency at the boundary (queue + worker hop) in exchange for reliability, horizontal scalability, and clean separation between request handling and long-running execution. For agents that complete in under two seconds, you can collapse the queue layer and run the graph inline in the FastAPI handler. For anything involving multiple LLM calls, tool use, or steps with real-world I/O, the queue-and-worker pattern is not optional — it’s the difference between a system that works and one that works until your first LLM timeout.
Setting Up LangGraph with Redis Checkpointing
LangGraph ships with a first-party Redis checkpointer via langgraph-checkpoint-redis. Setting it up correctly requires understanding the threading model LangGraph uses.
from langgraph.checkpoint.redis import RedisSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
import redis
# Define agent state schema
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
tool_calls_made: int
task_complete: bool
final_result: str | None
# Initialize Redis checkpointer
redis_client = redis.Redis(
host="redis",
port=6379,
decode_responses=False, # Checkpointer needs bytes
socket_connect_timeout=5,
socket_timeout=5,
retry_on_timeout=True,
)
checkpointer = RedisSaver(redis_client)
checkpointer.setup() # Creates required Redis data structures
Two things matter here that most examples skip. First, decode_responses=False — the checkpointer serializes state as bytes, and setting this to True will cause silent deserialization failures that are painful to debug. Second, checkpointer.setup() is idempotent but required on first run; it creates the stream and sorted set structures LangGraph uses internally. Run it at application startup, not lazily.
Defining the Graph with Production-Grade Nodes
A production graph needs more than just LLM calls. Each node should handle its own errors, respect timeouts, and emit enough telemetry to reconstruct what happened post-mortem.
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage
import asyncio
llm = ChatAnthropic(model="claude-sonnet-4-6", timeout=30)
async def reasoning_node(state: AgentState) -> dict:
"""Core LLM reasoning step with retry logic."""
try:
response = await llm.ainvoke(state["messages"])
return {
"messages": [response],
"tool_calls_made": state["tool_calls_made"],
}
except asyncio.TimeoutError:
# Return a recoverable error message rather than raising
# so LangGraph can checkpoint the failure state
error_msg = AIMessage(content="[TIMEOUT: LLM call exceeded 30s]")
return {"messages": [error_msg]}
async def tool_execution_node(state: AgentState) -> dict:
"""Execute tool calls from the last AI message."""
last_message = state["messages"][-1]
if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
return {}
# Execute tools with bounded concurrency
results = await execute_tools_with_timeout(
last_message.tool_calls,
timeout_per_tool=10.0,
max_concurrent=3,
)
return {
"messages": results,
"tool_calls_made": state["tool_calls_made"] + len(results),
}
def should_continue(state: AgentState) -> str:
"""Route to next node based on state."""
last_message = state["messages"][-1]
if state["task_complete"]:
return "finalize"
if state["tool_calls_made"] >= 10:
return "finalize" # Hard limit: prevent infinite tool loops
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "execute_tools"
return "finalize"
# Build the graph
graph_builder = StateGraph(AgentState)
graph_builder.add_node("reason", reasoning_node)
graph_builder.add_node("execute_tools", tool_execution_node)
graph_builder.add_node("finalize", finalize_node)
graph_builder.set_entry_point("reason")
graph_builder.add_conditional_edges("reason", should_continue)
graph_builder.add_edge("execute_tools", "reason")
graph_builder.add_edge("finalize", END)
agent_graph = graph_builder.compile(checkpointer=checkpointer)
The tool_calls_made >= 10 guard is not arbitrary. In production, the most common cause of runaway agent cost is an LLM that enters a tool-call loop — calling the same tool repeatedly because it misinterprets results. A hard ceiling on tool invocations per thread is cheap insurance. Set it based on your task’s expected complexity, not on what feels like a lot.
FastAPI Integration: Async Execution and Streaming
The FastAPI layer has two responsibilities: accepting agent execution requests and streaming results back to clients. These should be separate endpoints — mixing them into a single long-poll creates backpressure problems at scale.
The Submission Endpoint
from fastapi import FastAPI, BackgroundTasks, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import uuid
import json
app = FastAPI()
class AgentRequest(BaseModel):
task: str
thread_id: str | None = None # None = new thread, str = resume existing
class AgentResponse(BaseModel):
thread_id: str
stream_url: str
@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest, background_tasks: BackgroundTasks):
thread_id = request.thread_id or str(uuid.uuid4())
# Validate thread ownership here (omitted for brevity)
# In production: verify the requesting user owns this thread_id
background_tasks.add_task(
execute_agent_task,
thread_id=thread_id,
task=request.task,
)
return AgentResponse(
thread_id=thread_id,
stream_url=f"/agent/stream/{thread_id}",
)
The Streaming Endpoint
Server-Sent Events are the right primitive here. WebSockets are bidirectional and stateful — appropriate for interactive agents, but overkill for the common case of “submit task, receive results.” SSE gives you streaming over plain HTTP with built-in reconnection semantics.
import asyncio
from redis.asyncio import Redis as AsyncRedis
async_redis = AsyncRedis(host="redis", port=6379, decode_responses=True)
@app.get("/agent/stream/{thread_id}")
async def stream_agent(thread_id: str):
async def event_generator():
pubsub = async_redis.pubsub()
await pubsub.subscribe(f"agent:events:{thread_id}")
try:
# Stream events until completion or timeout (5 minutes hard limit)
deadline = asyncio.get_event_loop().time() + 300
async for message in pubsub.listen():
if asyncio.get_event_loop().time() > deadline:
yield format_sse({"type": "timeout", "thread_id": thread_id})
break
if message["type"] != "message":
continue
data = json.loads(message["data"])
yield format_sse(data)
if data.get("type") in ("complete", "error"):
break
finally:
await pubsub.unsubscribe(f"agent:events:{thread_id}")
await pubsub.close()
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no", # Critical for nginx deployments
},
)
def format_sse(data: dict) -> str:
return f"data: {json.dumps(data)}\n\n"
The X-Accel-Buffering: no header is a frequent production gotcha. Without it, nginx buffers SSE responses and clients receive bursts of events instead of a real-time stream. This manifests as a UI that appears frozen and then suddenly updates — which looks like a bug in your agent when it’s actually a proxy misconfiguration.
State Resumption: The Feature That Changes Everything
One of LangGraph’s most powerful production features is thread-based resumption. When a client passes an existing thread_id, the agent resumes from the last checkpoint rather than starting over. This has significant implications for reliability and user experience.
async def execute_agent_task(thread_id: str, task: str):
config = {
"configurable": {
"thread_id": thread_id,
}
}
# Check if this thread has prior state
state = await agent_graph.aget_state(config)
if state.values:
# Resuming: append the new task as a human message
initial_input = {"messages": [HumanMessage(content=task)]}
else:
# New thread: initialize full state
initial_input = {
"messages": [HumanMessage(content=task)],
"tool_calls_made": 0,
"task_complete": False,
"final_result": None,
}
async for event in agent_graph.astream_events(initial_input, config=config, version="v2"):
# Publish each event to Redis pub/sub for the streaming endpoint
await publish_agent_event(thread_id, event)
This resumption pattern enables a class of UX that’s impossible with stateless agents: “continue from where we left off” interactions, long-running background tasks that clients can reconnect to after network interruptions, and human-in-the-loop workflows where execution pauses waiting for user input.
Redis Key Design for Production Scale
Ad-hoc Redis key naming leads to namespace collisions, uncontrolled memory growth, and impossible-to-debug state leakage. Define your key schema upfront:
agent:checkpoint:{thread_id} # LangGraph checkpoint data (managed by checkpointer)
agent:events:{thread_id} # Pub/sub channel for streaming
agent:result:{thread_id} # Final result, TTL 24h
agent:meta:{thread_id} # Thread metadata (user_id, created_at, status)
agent:active_threads # Sorted set of active thread_ids by start time
Set explicit TTLs on everything that isn’t managed by the checkpointer:
RESULT_TTL = 86400 # 24 hours
META_TTL = 604800 # 7 days
ACTIVE_THREAD_TTL = 3600 # 1 hour (renewed on activity)
Without TTLs, Redis memory grows unbounded. In a system running thousands of agent threads per day, checkpoint data accumulates fast. The LangGraph Redis checkpointer does not automatically expire old checkpoints — you need a cleanup job or explicit TTL policy.
Failure Modes and How to Handle Them
LLM Provider Timeouts Mid-Graph
The most common production failure. An LLM call times out after 30 seconds, the node raises an exception, and the execution thread dies. Without checkpointing, all prior work is lost.
With Redis checkpointing, the state at the last completed node is preserved. The correct recovery pattern is not to auto-retry the failed thread immediately (which risks duplicate tool calls if the timeout was on the LLM side, not the network), but to surface the failure to a monitoring system and allow controlled retry with idempotency checks.
Redis Connection Loss During Checkpoint Write
If Redis becomes unavailable mid-write, LangGraph will raise a connection error. This is preferable to silent data corruption, but requires your worker to handle it gracefully. Use connection pooling with health checks and implement exponential backoff on reconnection attempts. Do not silently swallow Redis errors — they represent state loss.
Worker Process Crash During Node Execution
The last checkpoint written before the crash is the recovery point. This means in-flight work within a node is lost. Keep nodes granular — a node that does five things should be five nodes, so that crashes within a node lose at most one step of work rather than five.
Observability: What You Need in Production
A production agent system needs three observability signals, and most teams implement only the first.
Trace-level telemetry: Every LangGraph node execution, with duration, input/output token counts, and tool call outcomes. Use LangSmith or instrument with OpenTelemetry directly. Without this, debugging a failed agent run is archaeology.
Thread-level metrics: Active threads, checkpoint write latency, queue depth, and error rates broken down by failure type (LLM timeout vs. tool failure vs. Redis error). These tell you whether the system is healthy, not just whether individual requests succeeded.
Cost attribution: Token usage per thread, per user, per workflow type. AI agent costs are non-uniform and can spike unexpectedly. Implement per-thread token accounting at the node level using LangChain callbacks, and emit these metrics to your observability platform.
Deployment: Container Configuration That Actually Matters
# docker-compose.yml excerpt for production-like local deployment
services:
api:
build: .
command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
environment:
- REDIS_URL=redis://redis:6379
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
depends_on:
redis:
condition: service_healthy
worker:
build: .
command: python -m app.worker
environment:
- REDIS_URL=redis://redis:6379
- WORKER_CONCURRENCY=10
deploy:
replicas: 3
redis:
image: redis:7-alpine
command: redis-server --appendonly yes --maxmemory 2gb --maxmemory-policy allkeys-lru
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
The Redis maxmemory-policy allkeys-lru setting is a deliberate choice: when Redis approaches its memory limit, it evicts the least-recently-used keys. This means old checkpoint data is evicted before active thread state, which is the correct priority ordering. The alternative (noeviction) causes Redis to start returning errors when memory is full — which will crash your agents mid-execution.
When This Stack Is and Isn’t the Right Choice
This architecture is appropriate for agents that run multi-step tasks (more than two LLM calls), need to handle concurrent users at meaningful scale, require durability across process restarts, or involve human-in-the-loop interaction patterns.
It is overengineered for simple single-turn agents, internal tools with low concurrency requirements, or proof-of-concept systems where operational complexity is a liability rather than an asset. For those cases, LangGraph with an in-memory checkpointer and a synchronous FastAPI endpoint is simpler, correct, and easier to reason about.
The harness should fit the agent. This stack is a production harness — apply it when production requirements justify it.
What to Build Next
If you’ve implemented this stack and it’s running cleanly, the next layer of reliability comes from three additions:
Distributed tracing via OpenTelemetry — instrument your FastAPI endpoints, Redis operations, and LangGraph node executions with a shared trace context so you can follow a single agent request across all system components.
Thread prioritization — implement Redis sorted sets to give different priority weights to different request types, so critical agent tasks aren’t queued behind batch jobs.
Checkpoint compaction — implement periodic summarization of long thread histories to prevent checkpoint size from growing unbounded, reducing checkpoint write latency and Redis memory consumption.
Each of these is an independent concern you can add without restructuring the core architecture. That modularity is a feature of this design, not an accident.
Production AI agents are not complicated because of the LLM calls. They are complicated because of everything around the LLM calls: state that needs to survive failures, requests that need to be served concurrently, results that need to stream in real time, and costs that need to be tracked at granular resolution. LangGraph, Redis, and FastAPI each handle one part of that surface area cleanly. The discipline is in keeping those boundaries sharp.
Dr. Sarah Chen is a Principal Engineer at harness-engineering.ai, focused on production patterns and reliability architecture for AI agent systems. If you’re designing a production agent harness and want to go deeper on checkpointing strategies, state schema design, or async execution patterns, the architecture patterns library covers these topics in detail.