Building Production Agent Infrastructure with LangGraph

Understanding LangGraph's State Machine Model

At its core, LangGraph is a state machine framework built specifically for agents. Instead of thinking about agents as black boxes, you think about them as transformations of state. This is a powerful mental model because it makes your agent behavior predictable and reproducible.

The key insight that separates production agents from toy projects: everything your agent does must be serializable. Every decision point, every intermediate result, every piece of context must be reducible to data that can be stored and replayed. This constraint feels restrictive at first, but it becomes your superpower in production. When something breaks, you can replay exactly what happened. When you need to debug, you can step through the state machine. When you want to resume interrupted work, you just load the state and continue.

The TypedDict State Schema

Every LangGraph application starts with defining your state. This is where the magic begins - you're declaring what information flows through your agent system.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
 
class AgentState(TypedDict):
    """State schema for our research agent."""
    messages: list  # Conversation history
    query: str  # Original research query
    research_results: list  # Accumulated findings
    current_source: str  # What we're reading now
    final_report: str  # The completed analysis
    iteration_count: int  # Track recursion depth

Why a TypedDict? Because it forces you to be explicit about what data your agent needs. When you define state this way, you gain:

• Type safety: IDEs can autocomplete your state fields
• Serialization clarity: Everything that flows through your agent is serializable
• Debugging aid: You can see exactly what data is available at each step
• Checkpointing ability: State becomes reproducible and resumable

Think of the state schema as a contract. It says "this is everything the agent knows at any given moment." Nothing is hidden in closures or class variables. This explicitness is what makes production agents robust.

Node Functions as State Transformers

Now you define what happens at each node. A node is just a function that takes state and returns updated state.

def research_node(state: AgentState) -> AgentState:
    """Search for information relevant to the query."""
    query = state["query"]
 
    # Simulate searching (in production, this calls your search API)
    results = search_web(query)
 
    # Accumulate results
    state["research_results"].extend(results)
    state["current_source"] = results[0]["url"] if results else "none"
 
    return state
 
def analyze_node(state: AgentState) -> AgentState:
    """Synthesize research into a report."""
    results = state["research_results"]
    messages = state["messages"]
 
    # Use the LLM to analyze
    analysis_prompt = f"""
    Based on these research results: {results}
    Generate a comprehensive report answering: {state['query']}
    """
 
    response = llm.invoke(analysis_prompt)
    state["final_report"] = response.content
    state["messages"].append({"role": "assistant", "content": response.content})
 
    return state
 
def decide_more_research(state: AgentState) -> Literal["research_node", "analyze_node", END]:
    """Should we research more or finalize?"""
    iteration_count = state.get("iteration_count", 0)
 
    # Safety check: don't loop forever
    if iteration_count >= 3:
        return END
 
    # Check if we have enough data
    if len(state["research_results"]) >= 5:
        return "analyze_node"
 
    # Do another search iteration
    state["iteration_count"] = iteration_count + 1
    return "research_node"

This is where it clicks: each node is a pure function that takes state and returns modified state. No hidden side effects. No mysterious internal variables. Just data in, transformation, data out.

Conditional Edges for Branching

Real agents make decisions. They don't always follow a linear path. Conditional edges let your graph branch based on state. This is how you implement agent reasoning - the system examines the current state and decides what to do next based on logic you define.

from langgraph.graph import StateGraph, END
 
graph_builder = StateGraph(AgentState)
 
# Add nodes
graph_builder.add_node("research", research_node)
graph_builder.add_node("analyze", analyze_node)
 
# Set entry point
graph_builder.set_entry_point("research")
 
# Add conditional edge (branching logic)
graph_builder.add_conditional_edges(
    source="research",
    path=decide_more_research,
    conditional_edge_mapping={
        "research_node": "research",  # Loop back
        "analyze_node": "analyze",    # Move to analysis
        END: END                        # Finish
    }
)
 
# Add final edge
graph_builder.add_edge("analyze", END)

The add_conditional_edges method is where your agent's logic lives. It evaluates the current state and decides what happens next. This is how you implement agent reasoning in code. For production systems, this conditional routing is where you add your domain knowledge.

Graph Compilation with Checkpointing Backend

Now here's the production part: compilation. You don't just run your graph loose. You compile it with a checkpoint backend.

from langgraph.checkpoint.postgres import PostgresSaver
 
# Initialize PostgreSQL checkpoint backend
conn_string = "postgresql://user:password@localhost/langgraph_db"
checkpointer = PostgresSaver.from_conn_string(conn_string)
 
# Compile the graph with checkpointing
graph = graph_builder.compile(checkpointer=checkpointer)

What does checkpointing do? Every time your agent reaches a node, the entire state is saved to the database. If something crashes, you can resume from the last checkpoint. This is non-negotiable for production. Without checkpointing, if your agent crashes halfway through processing, you lose all work and context. With checkpointing, you resume from the exact point where it stopped.

# Run the agent with checkpointing
result = graph.invoke(
    {"messages": [], "query": "What is quantum computing?",
     "research_results": [], "current_source": "", "final_report": "",
     "iteration_count": 0},
    config={"configurable": {"thread_id": "research-session-001"}}
)
 
print(result["final_report"])

The thread_id is crucial. It ties all checkpoints together, letting you replay the exact execution path if needed for debugging. Think of it as a unique identifier for an agent execution session.

Persistence and Resumability

Checkpointing is nice, but persistence and resumability is where LangGraph becomes enterprise-grade. Let's dig into how you actually make an agent that survives failures.

Multi-Level Checkpointing Strategy

In production, you want multiple layers:

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.memory import MemorySaver
import os
 
class CheckpointingStrategy:
    """Manage checkpointing across environments."""
 
    @staticmethod
    def get_checkpointer():
        """Select checkpointer based on environment."""
        if os.getenv("ENV") == "production":
            # Production: durable PostgreSQL
            conn = os.getenv("DATABASE_URL")
            return PostgresSaver.from_conn_string(conn)
        else:
            # Development: in-memory for speed
            return MemorySaver()
 
checkpointer = CheckpointingStrategy.get_checkpointer()
graph = graph_builder.compile(checkpointer=checkpointer)

Why two levels? Because local development doesn't need durability, but production does. PostgreSQL is your guarantee that no work is lost. During development, in-memory checkpointing lets you iterate fast without database overhead.

Human-in-the-Loop Interruption Patterns

Sometimes you need to pause execution and ask a human what to do. LangGraph makes this elegant:

def needs_human_approval(state: AgentState) -> bool:
    """Check if we need human validation."""
    # Maybe we found conflicting information
    if len(state["research_results"]) > 0:
        have_conflicts = any(r["conflicting"] for r in state["research_results"])
        return have_conflicts
    return False
 
graph_builder.add_node(
    "human_review",
    lambda state: state  # Placeholder—state passes through
)
 
graph_builder.add_conditional_edges(
    source="research",
    path=lambda state: "human_review" if needs_human_approval(state) else "analyze",
    conditional_edge_mapping={
        "human_review": "human_review",
        "analyze": "analyze"
    }
)
 
# Now interrupt BEFORE the human review node
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # Pause here
)

When execution hits the interrupt point, the thread is suspended with full state preserved. Your API can wait for human input:

# In your API handler
for event in graph.stream(input_data, config):
    pass  # Let it run until interrupt
 
# Get the current state
state = graph.get_state(config)
print(f"Interrupted at: {state.next}")  # Shows the next node
print(f"Current data: {state.values}")   # Show the agent state
 
# Human reviews and decides
human_decision = get_human_input()
 
# Resume from the interrupt point
result = graph.invoke(
    {"messages": state.values["messages"], ...},
    config=config
)

This is how you build agents that respect human oversight. No more fire-and-forget systems that go rogue.

State Time-Travel for Debugging

Here's a debugging superpower: replay any point in execution history. This is invaluable when something goes wrong.

def get_state_history(thread_id: str, checkpointer):
    """Retrieve all states for a thread."""
    history = []
 
    # Query the checkpointer
    states = checkpointer.list_tuples(
        config={"configurable": {"thread_id": thread_id}}
    )
 
    for state in states:
        history.append({
            "checkpoint_id": state.checkpoint_id,
            "state": state.values,
            "timestamp": state.metadata.get("ts_seconds")
        })
 
    return history
 
# In your debugging CLI
history = get_state_history("research-session-001", checkpointer)
 
for i, checkpoint in enumerate(history):
    print(f"\nCheckpoint {i}:")
    print(f"  Research results: {len(checkpoint['state']['research_results'])}")
    print(f"  Current source: {checkpoint['state']['current_source']}")
 
# Jump back to a specific checkpoint and replay forward
config = {
    "configurable": {
        "thread_id": "research-session-001",
        "checkpoint_id": history[2]["checkpoint_id"]  # Go back to checkpoint 2
    }
}
 
# Resume from there
result = graph.invoke({}, config=config)

This is invaluable when something goes wrong. You can see exactly where the agent went off track and replay from any point.

Persistence and resumability are not optional features in production agent systems; they are fundamental requirements. When an agent is working on a complex task that takes minutes or hours, failures are inevitable. A network timeout. An external service becomes temporarily unavailable. A model behaves unexpectedly and the agent gets stuck in a loop. Without persistence, when the agent fails, you lose all the work it has done so far. You have to start over from the beginning, wasting computation and money. With proper persistence, you save the agent's state after each step. If it fails, you resume from where it left off.

LangGraph's checkpointing system enables this naturally. Each step in the agent's execution is a discrete atomic unit that can be checkpointed to persistent storage. If the agent fails at step ten, you can resume at step ten without re-executing steps one through nine. This is transformative for reliability. It means you can safely run long-running agents and recover from transient failures automatically.

The multi-level checkpointing strategy is particularly clever. You have checkpoints at different granularities: after each LLM call, after each tool invocation, after each major decision. This gives you flexibility in recovery strategies. Some failures might be recoverable by retrying just the last tool call. Others might require backing up further and taking a different path. The checkpoints give you options.

Human-in-the-loop interruption is another critical feature for production systems. Sometimes an agent gets stuck or makes a decision that a human needs to review. With interruption support, you can pause the agent, have a human review its state and decisions, make corrections if needed, and then resume. This is essential for high-stakes applications where you can't risk an agent making a bad decision autonomously. The checkpoint system makes this feasible because pausing and resuming is built into the architecture.

State time-travel for debugging is a capability that initially sounds like a luxury but becomes essential as agents grow more complex. You can replay an agent's execution and jump to any point in time to see what the state was, what decisions the agent made, and why. This turns debugging from a guessing game into an engineering task. You can understand exactly what went wrong and fix it systematically.

Multi-Agent Orchestration

Single agents are useful, but production systems need multiple specialized agents working together. LangGraph gives you patterns for this.

The Supervisor Pattern

Imagine you have specialist agents: a research agent, an analysis agent, and a writing agent. A supervisor LLM routes between them.

from typing import Literal
 
class OrchestratorState(TypedDict):
    """State for multi-agent orchestration."""
    task: str
    messages: list
    selected_agent: Literal["researcher", "analyzer", "writer", "end"]
    agent_outputs: dict  # Each agent's contribution
 
def supervisor_node(state: OrchestratorState) -> OrchestratorState:
    """Route to the appropriate agent."""
    messages = state["messages"]
 
    routing_prompt = f"""
    You are a supervisor orchestrating three specialist agents:
    - researcher: Finds information
    - analyzer: Synthesizes findings
    - writer: Creates final output
 
    Task: {state["task"]}
 
    Which agent should work next? Respond with ONLY: researcher, analyzer, writer, or end
    """
 
    response = llm.invoke(routing_prompt)
    next_agent = response.content.strip().lower()
 
    # Validate response
    valid_agents = ["researcher", "analyzer", "writer", "end"]
    if next_agent not in valid_agents:
        next_agent = "end"
 
    state["selected_agent"] = next_agent
    state["messages"].append({
        "role": "supervisor",
        "content": f"Routing to: {next_agent}"
    })
 
    return state
 
def researcher_node(state: OrchestratorState) -> OrchestratorState:
    """Research specialist agent."""
    results = search_and_summarize(state["task"])
    state["agent_outputs"]["research"] = results
    state["messages"].append({"role": "researcher", "content": results})
    return state
 
def analyzer_node(state: OrchestratorState) -> OrchestratorState:
    """Analysis specialist agent."""
    research = state["agent_outputs"].get("research", "")
    analysis = analyze_findings(research, state["task"])
    state["agent_outputs"]["analysis"] = analysis
    state["messages"].append({"role": "analyzer", "content": analysis})
    return state
 
def writer_node(state: OrchestratorState) -> OrchestratorState:
    """Writing specialist agent."""
    analysis = state["agent_outputs"].get("analysis", "")
    final_output = compose_report(analysis)
    state["agent_outputs"]["final"] = final_output
    state["messages"].append({"role": "writer", "content": final_output})
    return state
 
# Build the graph
orchestrator = StateGraph(OrchestratorState)
orchestrator.add_node("supervisor", supervisor_node)
orchestrator.add_node("researcher", researcher_node)
orchestrator.add_node("analyzer", analyzer_node)
orchestrator.add_node("writer", writer_node)
 
orchestrator.set_entry_point("supervisor")
 
# Conditional routing from supervisor
orchestrator.add_conditional_edges(
    "supervisor",
    lambda state: state["selected_agent"],
    {
        "researcher": "researcher",
        "analyzer": "analyzer",
        "writer": "writer",
        "end": END
    }
)
 
# Each agent routes back to supervisor for next decision
for agent in ["researcher", "analyzer", "writer"]:
    orchestrator.add_edge(agent, "supervisor")
 
compiled_orchestrator = orchestrator.compile(checkpointer=checkpointer)

This pattern is powerful because the supervisor can make intelligent routing decisions. If you need research, it sends work to the researcher. If you need synthesis, it routes to the analyzer. Each agent stays focused on its specialty.

Subgraph Composition

What if your researcher itself is complex? You can compose graphs within graphs:

# Build the research subgraph separately
research_graph = create_research_subgraph()
 
# Embed it in the larger orchestrator
orchestrator.add_node("researcher_subgraph", research_graph.invoke)
 
# Now the orchestrator can call the research subgraph as a single node

Subgraphs let you organize complexity. A large system might have dozens of subgraphs, each handling a domain-specific workflow.

Production Deployment Patterns

Okay, you've got your agent graph working locally. Now it's time to deploy. This is where things get real.

Architecture Overview

Here's what a production deployment-production-inference-deployment) looks like:

┌─────────────────────────────────────────────────────────┐
│                    API Gateway                           │
│              (FastAPI, authentication, routing)          │
└────────────────────┬────────────────────────────────────┘
                     │
    ┌────────────────┼────────────────┐
    │                │                │
┌───▼──────┐  ┌──────▼──────┐  ┌──────▼──────┐
│ Agent    │  │ Agent       │  │ Agent       │
│ Worker 1 │  │ Worker 2    │  │ Worker N    │
└───┬──────┘  └──────┬──────┘  └──────┬──────┘
    │                │                │
    └────────────────┼────────────────┘
                     │
         ┌───────────┼───────────┐
         │           │           │
    ┌────▼───┐  ┌────▼───┐  ┌───▼────┐
    │ Redis  │  │ Postgres│  │ Message│
    │ Cache  │  │ State   │  │ Queue  │
    └────────┘  └─────────┘  └────────┘

Multiple workers run the same agent graph. Checkpointing happens to PostgreSQL. A message queue (like Redis Streams) distributes work.

FastAPI Server with Streaming

Let's build the API layer:

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
import asyncio
import json
 
app = FastAPI()
 
@app.post("/agent/start")
async def start_agent(task: dict):
    """Start a new agent execution."""
    thread_id = generate_uuid()
 
    # Queue the task
    await queue.add_job(
        graph_id="orchestrator",
        input_data=task,
        thread_id=thread_id
    )
 
    return {"thread_id": thread_id, "status": "queued"}
 
@app.get("/agent/{thread_id}/stream")
async def stream_agent(thread_id: str):
    """Stream agent progress in real-time."""
 
    async def event_generator():
        config = {"configurable": {"thread_id": thread_id}}
 
        # Stream events from the graph
        for event in graph.stream({}, config=config):
            # Convert event to JSON
            event_json = json.dumps({
                "type": event.get("type"),
                "node": event.get("node"),
                "state": serialize_state(event.get("state", {}))
            })
 
            # Send as SSE
            yield f"data: {event_json}\n\n"
            await asyncio.sleep(0.1)  # Small delay
 
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )
 
@app.get("/agent/{thread_id}/status")
async def get_status(thread_id: str):
    """Get current state without streaming."""
    config = {"configurable": {"thread_id": thread_id}}
    state = graph.get_state(config)
 
    return {
        "thread_id": thread_id,
        "status": "running" if state.next else "complete",
        "current_node": state.next[0] if state.next else None,
        "state": serialize_state(state.values)
    }
 
@app.post("/agent/{thread_id}/resume")
async def resume_agent(thread_id: str, decision: dict):
    """Resume a paused agent with human input."""
    config = {"configurable": {"thread_id": thread_id}}
 
    # Update state with human decision
    state = graph.get_state(config)
    updated_state = {**state.values, **decision}
 
    # Resume execution
    result = graph.invoke(updated_state, config=config)
 
    return {"status": "resumed", "result": result}

Notice the stream_agent endpoint? That's Server-Sent Events (SSE). Your frontend can listen and get real-time updates as the agent runs. No polling. No delays.

Background Job Execution

Long-running agents shouldn't block HTTP requests. Use background jobs:

from celery import Celery
from celery.result import AsyncResult
 
celery_app = Celery("agent_tasks")
celery_app.conf.broker_url = "redis://localhost:6379"
 
@celery_app.task
def run_agent_task(thread_id: str, task_data: dict):
    """Run agent in background."""
    config = {"configurable": {"thread_id": thread_id}}
 
    try:
        result = graph.invoke(task_data, config=config)
        # Store result somewhere (database, cache, S3)
        save_result(thread_id, result)
    except Exception as e:
        log_error(thread_id, str(e))
        raise
 
@app.post("/agent/start-background")
async def start_background(task: dict, background_tasks: BackgroundTasks):
    """Start agent as background job."""
    thread_id = generate_uuid()
 
    # Queue with Celery
    job = run_agent_task.delay(thread_id, task)
 
    return {
        "thread_id": thread_id,
        "job_id": str(job.id),
        "status": "queued"
    }
 
@app.get("/agent/job/{job_id}")
async def check_job(job_id: str):
    """Check background job status."""
    result = AsyncResult(job_id, app=celery_app)
 
    return {
        "job_id": job_id,
        "status": result.state,
        "result": result.result if result.ready() else None
    }

This way, your API responds instantly while work happens in the background. Perfect for operations that might take minutes.

Horizontal Scaling with Kubernetes

Deploy multiple replicas:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-worker
spec:
  replicas: 3 # Start with 3 workers
  selector:
    matchLabels:
      app: agent-worker
  template:
    metadata:
      labels:
        app: agent-worker
    spec:
      containers:
        - name: agent-worker
          image: myregistry/agent-worker:latest
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: agent-secrets
                  key: database-url
            - name: REDIS_URL
              valueFrom:
                secretKeyRef:
                  name: agent-secrets
                  key: redis-url
          resources:
            requests:
              memory: "512Mi"
              cpu: "250m"
            limits:
              memory: "1Gi"
              cpu: "500m"
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 30
            periodSeconds: 10

Each worker runs the same graph code. PostgreSQL acts as the single source of truth for state. Kubernetes handles failover automatically.

The observability of agent systems is critical because agents are fundamentally opaque without proper instrumentation. You don't know what the agent is thinking. You don't know why it made a particular decision. You don't know if it's making progress or stuck in a loop. This opacity is dangerous in production. You need visibility into every step of the agent's execution.

LangSmith is designed specifically for this purpose. It traces every LLM call made by an agent, every tool invocation, and the full path through the state graph. You can see the exact prompt sent to the model, the exact response received, and how that response influenced the agent's subsequent decisions. This level of visibility is essential for debugging and optimization.

The trace ID tying everything together is crucial. When an agent fails or produces unexpected output, you can look up the trace ID and see the complete execution path. You can see every intermediate step, every decision, every tool call. This transforms debugging from a frustrating investigation into a straightforward analysis. You know exactly what happened and why.

Cost tracking for agents is more complex than cost tracking for simple LLM calls. An agent might make ten LLM calls to solve a single user request. You need to attribute the full cost of those ten calls to the user request. You need to break down the cost by tool, by model, by step type. This enables cost optimization. Maybe you discover that a particular tool is very expensive because it forces the agent to make many retries. You optimize the tool or its error handling. Or maybe you discover that a particular type of request causes the agent to take an inefficient path. You optimize the agent's reasoning or provide better context.

Production deployment patterns need to account for the fact that agents can take a long time to execute. A request from a user shouldn't block waiting for an agent to finish its work. Instead, agents should run as background jobs. The user submits a request, gets an acknowledgment, and then polls or subscribes to updates as the agent progresses. This requires a different API design than synchronous request-response. It requires job queuing, state tracking, and result persistence. But this pattern is necessary for scaling agents to production traffic.

Observability: LangSmith Integration

You can't operate blind in production. LangSmith is your window into what's happening.

Tracing Agent Execution Graphs

Set up LangSmith in your code:

import os
from langchain.callbacks.tracers import LangChainTracer
 
# Enable LangSmith tracing
os.environ["LANGCHAIN_API_KEY"] = "your-key"
os.environ["LANGCHAIN_PROJECT"] = "agent-production"
 
# All LangChain calls are automatically traced
llm = ChatOpenAI(model="gpt-4")
 
# Traces appear in LangSmith dashboard automatically
result = graph.invoke(input_data, config=config)

When you invoke the graph, every LLM call, every node execution, every state change is recorded. You see latency for each step, token usage, error traces.

Scaling agent systems to handle concurrent requests introduces additional complexity. You cannot just increase the number of LLM API calls proportionally because you hit rate limits, quota constraints, and cost limits. You need careful orchestration of agent execution. Some teams implement job queuing where agent requests are queued and processed in batches. This lets you control the rate at which agents consume expensive resources. Others implement prioritization where higher-priority requests get serviced first.

The interaction between agent parallelism and LLM rate limiting is particularly tricky. If you have fifty agents running concurrently and each makes three LLM calls during its execution, you're making one hundred and fifty concurrent API calls. If your rate limit is one thousand calls per minute, you've hit it. You need to implement backpressure where agents slow down or queue when approaching rate limits. The gateway pattern applies here too: a central orchestrator manages rate limits and distributes capacity fairly among agents.

Memory management is another practical concern. Agents accumulate state as they work. Long-running agents might accumulate gigabytes of state. If you're not careful about memory cleanup, you run out of memory. You need to implement periodic checkpointing to persistent storage and cleanup of old state. You need to monitor memory usage and alert when it grows unexpectedly.

Deploying agents to production requires thoughtful planning around visibility and debugging. When an agent makes a bad decision, you need to understand why. This requires comprehensive logging at every decision point. It requires tracing of the entire execution path. It requires capturing the prompts sent to the model and the responses received. Without this visibility, debugging agent misbehavior is nearly impossible.

Version control for agent code and configuration is essential. When you change the prompt or the agent logic, you want to be able to trace which version produced which output. You want to be able to roll back if you introduce a regression. You want to be able to compare two versions to understand what changed. This is standard software engineering practice that often gets overlooked in ML systems.

Putting It All Together: Complete Example

Let's build a real agent from scratch - a customer support classifier that routes tickets to specialist teams.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain.chat_models import ChatOpenAI
import json
 
# 1. Define state
class SupportTicketState(TypedDict):
    ticket_id: str
    customer_message: str
    category: str  # billing, technical, general
    priority: Literal["low", "medium", "high", "critical"]
    assigned_team: str
    response: str
    messages: list
    analysis_reasoning: str
 
# 2. Create node functions
def classify_ticket(state: SupportTicketState) -> SupportTicketState:
    """Classify the support ticket."""
    llm = ChatOpenAI(model="gpt-4")
 
    prompt = f"""
    Classify this support ticket:
    "{state['customer_message']}"
 
    Categories: billing, technical, general
    Priority: low, medium, high, critical
 
    Respond with JSON:
    {{"category": "...", "priority": "...", "reasoning": "..."}}
    """
 
    response = llm.invoke(prompt)
    data = json.loads(response.content)
 
    state["category"] = data["category"]
    state["priority"] = data["priority"]
    state["analysis_reasoning"] = data["reasoning"]
    state["messages"].append({
        "role": "classifier",
        "content": f"Classified as {data['category']} / {data['priority']}"
    })
 
    return state
 
def route_to_specialist(state: SupportTicketState) -> Literal["billing_team", "technical_team", "general_team"]:
    """Route based on category."""
    category_map = {
        "billing": "billing_team",
        "technical": "technical_team",
        "general": "general_team"
    }
    return category_map.get(state["category"], "general_team")
 
def billing_team_node(state: SupportTicketState) -> SupportTicketState:
    """Billing specialist response."""
    state["assigned_team"] = "billing"
    state["response"] = "We've escalated your billing inquiry to our finance team..."
    state["messages"].append({"role": "billing_team", "content": state["response"]})
    return state
 
def technical_team_node(state: SupportTicketState) -> SupportTicketState:
    """Technical specialist response."""
    state["assigned_team"] = "technical"
    state["response"] = "Our technical team is investigating your issue..."
    state["messages"].append({"role": "technical_team", "content": state["response"]})
    return state
 
def general_team_node(state: SupportTicketState) -> SupportTicketState:
    """General support response."""
    state["assigned_team"] = "general"
    state["response"] = "Thank you for reaching out. We're here to help..."
    state["messages"].append({"role": "general_team", "content": state["response"]})
    return state
 
# 3. Build the graph
graph_builder = StateGraph(SupportTicketState)
 
graph_builder.add_node("classify", classify_ticket)
graph_builder.add_node("billing_team", billing_team_node)
graph_builder.add_node("technical_team", technical_team_node)
graph_builder.add_node("general_team", general_team_node)
 
graph_builder.set_entry_point("classify")
 
graph_builder.add_conditional_edges(
    "classify",
    route_to_specialist,
    {
        "billing_team": "billing_team",
        "technical_team": "technical_team",
        "general_team": "general_team"
    }
)
 
for team in ["billing_team", "technical_team", "general_team"]:
    graph_builder.add_edge(team, END)
 
# 4. Compile with checkpointing
checkpointer = PostgresSaver.from_conn_string("postgresql://user:password@localhost/langgraph")
graph = graph_builder.compile(checkpointer=checkpointer)
 
# 5. Run it
result = graph.invoke(
    {
        "ticket_id": "TICKET-001",
        "customer_message": "I was charged twice for my subscription",
        "category": "",
        "priority": "medium",
        "assigned_team": "",
        "response": "",
        "messages": [],
        "analysis_reasoning": ""
    },
    config={"configurable": {"thread_id": "ticket-001"}}
)
 
print(f"Category: {result['category']}")
print(f"Assigned to: {result['assigned_team']}")
print(f"Response: {result['response']}")

Run this and you get perfect routing. Perfect traceability. Checkpointed at every step.

Why LangGraph for Production

Let me be direct: building agents without LangGraph is like building without version control. You can do it, but you're signing up for pain.

LangGraph gives you:

• State as a contract: Everything your agent needs is explicit, serializable, debuggable
• Deterministic execution: Same input + same checkpoints = same result
• Human oversight: Interruption patterns let humans stay in control
• Observability: LangSmith integration means you see everything
• Scalability: Stateless workers with centralized checkpointing scales horizontally
• Recoverability: Failed runs are resumable from any checkpoint

These aren't nice-to-haves. They're requirements. And LangGraph makes them standard.

The Trap of Stateful Agent Systems

Many teams start building agents with a simpler approach - just code the logic directly, no framework. The agent has instance variables, class state, and logic scattered across methods. It works great in notebooks. You spin it up, feed it a prompt, and it runs.

Then you try to deploy it. Now you need to handle the case where a request times out halfway through. Do you lose all the work? Do you restart from the beginning? What if the agent was mid-reasoning through a complex problem - do you force it to start over, wasting compute and time?

You discover you need persistence. You add a database. Now you're writing code to serialize your agent state to the database and reconstruct it on recovery. You've essentially reinvented what LangGraph does for you. Except your implementation is probably buggy, doesn't handle edge cases, and is tightly coupled to your specific agent design.

Or worse, you don't add persistence and just accept that long-running agents can fail and lose progress. Your users get frustrated when they ask the agent to help with a complex task and a network blip causes total failure. Your team spends time fighting production issues instead of building features.

LangGraph forces you to think about state as first-class from day one. You can't build a stateful agent system without confronting the serialization problem. And by confronting it early, you build systems that are actually reliable.

Debugging Agents is Impossible Without State Time-Travel

Here's a scenario that happens in production: an agent made a decision you disagree with. It routed a request incorrectly, or it chose the wrong tool, or it synthesized incorrect information. What do you do?

Without checkpoint history, you're stuck. You can log the final output, maybe log some intermediate states if you thought to add logging. But you can't replay the exact sequence of events and see where things went wrong. You certainly can't pause at the decision point and ask "why did you choose this?" and have the agent show you its reasoning.

With LangGraph's state checkpointing and time-travel capability, you can do exactly that. You load the specific checkpoint from the problem execution, examine the state, see what the agent knew at that decision point. You can even step through the execution manually, pause at any node, and examine what would happen if the agent made a different choice.

This debugging capability is invaluable when you're trying to understand why your agent is misbehaving in production. It transforms agent debugging from black-box guessing to white-box investigation.

The Interruption Pattern: Keeping Humans in Control

As AI systems become more capable, the pressure to fully automate everything grows. But wise teams resist this pressure. The most robust systems have humans in the loop at critical decision points.

LangGraph's interruption pattern makes this elegant. You don't need to redesign your entire system to add human approval. You specify which nodes are interruption points. Execution pauses automatically. Your system queries the human. The human provides input. Execution resumes. The agent continues with that human guidance.

This is how you build systems that scale human judgment. You automate 90% of the work, but the last 10% - the decisions that actually matter or have high stakes - require human approval. The agent does research, analysis, and synthesis. But whether to actually take action? That's human-decided.

When Not to Use LangGraph

That said, LangGraph isn't the right tool for every agent use case. Simple agents with straightforward logic and no persistence requirements don't need it. A chatbot that just wraps an LLM with some prompt engineering doesn't need state machines.

LangGraph shines when you have:

• Multi-step reasoning that you want to checkpoint
• Tool use that might fail and need recovery
• Conditional branching based on intermediate results
• Human-in-the-loop workflows
• Long-running agents that might get interrupted
• Complex orchestration of multiple LLM calls

If you have none of these, simpler tools might be better. But most production agents have at least some of these characteristics. And when they do, LangGraph is the right investment.

Operational Maturity and Observability

Running agents in production without visibility is just asking for trouble. You deploy an agent system, and three months later it's making systematically biased decisions in a way you don't immediately notice. By the time you find the problem, it's hurt real users.

LangGraph's integration with LangSmith creates a natural observability layer. Every run is visible. You can see execution traces, token usage, latency for each step. You can compare how different user requests flow through your agent. You can identify patterns in failures.

This observability is almost impossible to retrofit into ad-hoc agent systems. It requires thoughtful instrumentation, and you usually discover too late that you didn't log the right information. LangGraph logs everything automatically.

Scaling from Prototype to Production

Most agent projects start in a notebook or local script. The transition to production is where most projects fail or get massively rewritten.

With LangGraph, that transition is smooth. Your local development graph using MemorySaver works exactly the same way as production with PostgresSaver. You test with fast in-memory checkpointing, then switch the backend and deploy. The agent logic doesn't change.

You can gradually add features: first basic state management, then human interruption, then multi-agent orchestration, then observability. Each piece plugs in without requiring major refactoring.

** One consideration that often gets overlooked is the interaction between agent persistence and business logic. When you save an agent's state to resume later, you're making assumptions about the state being valid in the future. But business context might change. A user account might be suspended. A resource might be deleted. An API might be changed. When you resume an agent, you need to re-validate that the saved state is still meaningful. This requires defensive programming and proper error handling.

The observability infrastructure supporting agents needs to capture the right level of detail. Logging every single variable at every step creates too much data. Logging only final outcomes loses useful context. The sweet spot is logging at decision points, logging errors, and sampling the rest. This gives visibility into why the agent made decisions without drowning in data.

The user experience of waiting for agent execution is important. Many agents take minutes or hours to complete. Users shouldn't sit waiting. You need asynchronous execution where the user gets an immediate acknowledgment and then polls or subscribes for updates. Some teams use WebSockets to push updates as the agent progresses. Others use polling or email notifications. The choice depends on your application and user expectations.

The cost of agent execution can spiral quickly if not managed carefully. Every LLM call costs money. Every tool invocation might call external APIs which cost money. An agent that makes fifty LLM calls and uses ten different APIs might cost dollars to execute. For complex multi-agent orchestrations, the cost per execution can be high. You need visibility into this cost and limits to prevent runaway costs.

The interaction between agent execution and system load is important. If all agents run as fast as possible, they can overwhelm your infrastructure. You need rate limiting, queuing, and prioritization to manage load fairly. You might limit how many agents can run concurrently. You might limit how many LLM calls can be made per second. You might prioritize critical requests over background jobs. These controls are essential for stable operations.

The culture around agent systems should emphasize measurement and continuous improvement. Instead of assuming an agent works well, measure its behavior. Track success rates, latency, cost, and quality. Build dashboards showing agent performance over time. Analyze failure cases to understand what went wrong. Use this data to improve the agent. This measurement-driven approach makes agent systems more reliable and efficient over time.

Finally, remember that agents are tools for your users, not ends in themselves. The goal is to solve real problems for real users. An agent that optimizes perfectly for some metric you invented but doesn't actually help users is a failed agent. Keep user needs in mind as you design and optimize your agent systems. Let user feedback drive your priorities.