Building AI Agents with LangGraph: From Fundamentals to Production

A comprehensive guide to building reliable, stateful AI agents using LangGraph — the orchestration framework for agentic workflows.

Introduction

AI agents are LLMs equipped with tools, memory, and reasoning loops. LangGraph provides the orchestration layer to make them production-ready.

Why LangGraph?

Feature	LangChain Agents	LangGraph
State Management	Implicit, limited	Explicit, full control
Cycles/Loops	Difficult	Native support
Human-in-the-Loop	Basic	First-class
Debugging	Hard	Time-travel debugging
Persistence	Manual	Built-in checkpoints
Multi-Agent	Ad-hoc	Structured graphs

Core Concepts

1. State

The single source of truth for your agent:

# state.py
from typing import TypedDict, List, Annotated
from langgraph.graph import add_messages
from pydantic import BaseModel

class AgentState(TypedDict):
    # Messages with automatic reducer (appends)
    messages: Annotated[List[BaseModel], add_messages]

    # Custom fields
    user_id: str
    session_id: str
    tool_call_count: int
    current_plan: str
    retrieved_docs: List[dict]
    error_count: int

2. Nodes

Functions that process state and return updates:

# nodes.py
from langgraph.graph import END
from langchain_core.messages import AIMessage, ToolMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

async def agent_node(state: AgentState):
    """Main reasoning node."""
    response = await llm_with_tools.ainvoke(state["messages"])

    return {
        "messages": [response],
        "tool_call_count": state.get("tool_call_count", 0) + 
                          len(response.tool_calls) if response.tool_calls else 0
    }

async def tool_node(state: AgentState):
    """Execute tool calls."""
    last_message = state["messages"][-1]
    tool_outputs = []

    for tool_call in last_message.tool_calls:
        tool = get_tool_by_name(tool_call["name"])
        try:
            result = await tool.ainvoke(tool_call["args"])
            tool_outputs.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))
        except Exception as e:
            tool_outputs.append(ToolMessage(
                content=f"Error: {str(e)}",
                tool_call_id=tool_call["id"]
            ))

    return {"messages": tool_outputs}

def should_continue(state: AgentState) -> str:
    """Route: continue to tools or end."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

3. Edges & Conditional Routing

# graph.py
from langgraph.graph import StateGraph, START

def build_agent_graph():
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tool_node)
    workflow.add_node("validate", validation_node)

    # Edges
    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {"tools": "tools", END: END}
    )
    workflow.add_edge("tools", "agent")

    # Compile with checkpointing
    return workflow.compile(
        checkpointer=MemorySaver(),  # Or PostgresSaver for production
        interrupt_before=["tools"]  # Human-in-the-loop
    )

Building a Production Agent

Complete Example: Research Assistant

# research_agent.py
from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
import json
import sqlite3
from datetime import datetime

# ============ STATE ============
class ResearchState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    research_topic: str
    sub_questions: List[str]
    gathered_info: List[dict]
    current_step: str
    iteration: int
    max_iterations: int
    final_report: str

# ============ TOOLS ============
@tool
def web_search(query: str, num_results: int = 5) -> str:
    """Search the web for information."""
    # Implementation using your preferred search API
    # Return formatted results
    pass

@tool
def save_research_note(topic: str, content: str, source: str) -> str:
    """Save a research finding."""
    conn = sqlite3.connect("research.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY,
            topic TEXT, content TEXT, source TEXT, created_at TEXT
        )
    """)
    conn.execute(
        "INSERT INTO notes (topic, content, source, created_at) VALUES (?, ?, ?, ?)",
        (topic, content, source, datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()
    return f"Saved note for '{topic}'"

tools = [web_search, save_research_note]

# ============ NODES ============
llm = ChatOpenAI(model="gpt-4o", temperature=0)
planner_llm = llm.bind_tools(tools)
researcher_llm = llm.bind_tools([web_search])
writer_llm = llm.bind_tools([save_research_note])

async def planner(state: ResearchState):
    """Plan the research approach."""
    topic = state["research_topic"]

    prompt = f"""Break down this research topic into 3-5 specific sub-questions:
    Topic: {topic}

    Return as JSON: {{"sub_questions": ["q1", "q2", ...]}}"""

    response = await planner_llm.ainvoke([HumanMessage(content=prompt)])
    plan = json.loads(response.content)

    return {
        "sub_questions": plan["sub_questions"],
        "current_step": "researching",
        "iteration": 0,
        "gathered_info": []
    }

async def researcher(state: ResearchState):
    """Research each sub-question."""
    if state["iteration"] >= len(state["sub_questions"]):
        return {"current_step": "writing"}

    question = state["sub_questions"][state["iteration"]]

    prompt = f"""Research this question thoroughly using web search:
    Question: {question}

    Search for multiple sources and synthesize findings."""

    response = await researcher_llm.ainvoke([
        HumanMessage(content=prompt)
    ])

    # Execute tool calls
    tool_results = []
    for tool_call in response.tool_calls:
        if tool_call["name"] == "web_search":
            result = await web_search.ainvoke(tool_call["args"])
            tool_results.append({
                "question": question,
                "result": result,
                "source": "web_search"
            })

    return {
        "gathered_info": state["gathered_info"] + tool_results,
        "iteration": state["iteration"] + 1
    }

async def writer(state: ResearchState):
    """Compile final report."""
    prompt = f"""Write a comprehensive research report on: {state['research_topic']}

    Sub-questions and findings:
    {json.dumps(state['gathered_info'], indent=2)}

    Structure:
    1. Executive Summary
    2. Detailed Findings (per sub-question)
    3. Key Insights
    4. Sources
    5. Limitations"""

    response = await writer_llm.ainvoke([HumanMessage(content=prompt)])

    # Save final report
    await save_research_note.ainvoke({
        "topic": state["research_topic"],
        "content": response.content,
        "source": "final_report"
    })

    return {
        "final_report": response.content,
        "current_step": "complete"
    }

def route_research(state: ResearchState) -> Literal["researcher", "writer"]:
    return state["current_step"]

# ============ GRAPH ============
def build_research_agent():
    workflow = StateGraph(ResearchState)

    workflow.add_node("planner", planner)
    workflow.add_node("researcher", researcher)
    workflow.add_node("writer", writer)

    workflow.add_edge(START, "planner")
    workflow.add_edge("planner", "researcher")
    workflow.add_conditional_edges(
        "researcher",
        route_research,
        {"researcher": "researcher", "writer": "writer"}
    )
    workflow.add_edge("writer", END)

    # Persistence
    checkpointer = SqliteSaver.from_conn_string("sqlite:///research_agent.db")

    return workflow.compile(checkpointer=checkpointer)

# ============ USAGE ============
async def run_research(topic: str, thread_id: str):
    agent = build_research_agent()

    config = {"configurable": {"thread_id": thread_id}}

    initial_state = {
        "messages": [HumanMessage(content=f"Research: {topic}")],
        "research_topic": topic,
        "max_iterations": 5
    }

    # Stream for progress
    async for chunk in agent.astream(initial_state, config):
        for node, update in chunk.items():
            print(f"[{node}] {update.get('current_step', '')}")
            if update.get("final_report"):
                print("\n=== FINAL REPORT ===")
                print(update["final_report"])

# Run
# asyncio.run(run_research("Impact of AI on software development jobs", "thread-123"))

Advanced Patterns

1. Multi-Agent Collaboration

# multi_agent.py
from langgraph.graph import StateGraph
from typing import Literal

class MultiAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    next_agent: str
    task: str
    results: dict

def supervisor(state: MultiAgentState) -> Literal["researcher", "coder", "reviewer", END]:
    """Route to appropriate specialist."""
    last_msg = state["messages"][-1].content.lower()

    if "research" in last_msg or "find" in last_msg:
        return "researcher"
    elif "code" in last_msg or "implement" in last_msg:
        return "coder"
    elif "review" in last_msg or "check" in last_msg:
        return "reviewer"
    return END

# Each specialist is a subgraph
researcher_graph = build_research_subgraph()
coder_graph = build_coder_subgraph()
reviewer_graph = build_reviewer_subgraph()

workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_graph)
workflow.add_node("coder", coder_graph)
workflow.add_node("reviewer", reviewer_graph)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", supervisor)
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("coder", "supervisor")
workflow.add_edge("reviewer", "supervisor")

2. Human-in-the-Loop

# human_in_loop.py
from langgraph.types import interrupt, Command

async def sensitive_action(state: AgentState):
    """Action requiring human approval."""
    action = state.get("pending_action")

    # Interrupt and wait for human
    human_response = interrupt({
        "action": action,
        "question": f"Approve {action['name']} with params {action['params']}?"
    })

    if human_response.get("approved"):
        result = await execute_action(action)
        return {"messages": [ToolMessage(content=result, tool_call_id=action["id"])]}
    else:
        return {"messages": [ToolMessage(
            content=f"Rejected: {human_response.get('reason')}",
            tool_call_id=action["id"]
        )]}

# Resume after human input
config = {"configurable": {"thread_id": "thread-1"}}
agent.invoke(Command(resume={"approved": True, "reason": "Looks good"}), config)

3. Time-Travel Debugging

# debugging.py
# Get state history
history = list(agent.get_state_history(config))

for state_snapshot in history:
    print(f"Step: {state_snapshot.metadata['step']}")
    print(f"Messages: {len(state_snapshot.values['messages'])}")
    print(f"Next: {state_snapshot.next}")
    print("---")

# Replay from specific point
past_config = agent.get_state({"configurable": {"thread_id": "thread-1", "checkpoint_id": "checkpoint-5"}})
agent.invoke(new_input, past_config)  # Branches from that point

4. Structured Output with Validation

# structured_output.py
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field, field_validator

class CodeReview(BaseModel):
    file: str
    issues: List[dict] = Field(default_factory=list)
    suggestions: List[str] = Field(default_factory=list)
    score: int = Field(ge=0, le=100)

    @field_validator("score")
    @classmethod
    def validate_score(cls, v):
        if not 0 <= v <= 100:
            raise ValueError("Score must be 0-100")
        return v

parser = PydanticOutputParser(pydantic_object=CodeReview)

review_prompt = f"""Review this code and output JSON:
{parser.get_format_instructions()}

Code:
{{code}}"""

async def code_reviewer(state):
    code = state["code_to_review"]
    response = await llm.ainvoke(review_prompt.format(code=code))

    try:
        review = parser.parse(response.content)
        return {"review": review.model_dump()}
    except Exception as e:
        # Retry with error feedback
        return {"error": str(e), "retry": True}

Production Checklist

Reliability

• [ ] Checkpointing: Use PostgresSaver or RedisSaver for production
• [ ] Error handling: Wrap nodes in try/catch, add retry logic
• [ ] Timeouts: Set timeout on LLM calls and tool executions
• [ ] Rate limiting: Implement token bucket for external APIs
• [ ] Circuit breakers: Fail fast when dependencies are down

Observability

• [ ] LangSmith tracing: os.environ["LANGCHAIN_TRACING_V2"] = "true"
• [ ] Custom metrics: Log latency, token usage, error rates
• [ ] State logging: Periodic state snapshots for debugging

# monitoring.py
from langsmith import traceable

@traceable(name="agent_step")
async def monitored_node(state: AgentState):
    start = time.time()
    result = await agent_node(state)
    duration = time.time() - start

    # Custom metrics
    log_metric("node_latency", duration, tags={"node": "agent"})
    log_metric("token_usage", result.get("token_count", 0))

    return result

Scaling

• [ ] Horizontal scaling: Stateless nodes + external checkpointer
• [ ] Queue-based: Use Celery/RQ for async node execution
• [ ] Caching: Cache LLM responses for repeated queries

# caching.py
import hashlib
import redis

redis_client = redis.Redis(decode_responses=True)

async def cached_llm_call(prompt: str, model: str) -> str:
    cache_key = f"llm:{model}:{hashlib.md5(prompt.encode()).hexdigest()}"

    cached = redis_client.get(cache_key)
    if cached:
        return cached

    response = await llm.ainvoke(prompt)
    redis_client.setex(cache_key, 3600, response.content)  # 1hr TTL
    return response.content

Deployment

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Install LangGraph
RUN pip install langgraph langchain-openai langchain-core

# Copy app
COPY . .

# Run
CMD ["python", "server.py"]

FastAPI Server

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.sqlite import SqliteSaver
import uvicorn

app = FastAPI()
agent = build_research_agent()
checkpointer = SqliteSaver.from_conn_string("sqlite:///checkpoints.db")

class ChatRequest(BaseModel):
    message: str
    thread_id: str

class ChatResponse(BaseModel):
    response: str
    thread_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}

    try:
        result = await agent.ainvoke(
            {"messages": [HumanMessage(content=request.message)]},
            config
        )

        last_msg = result["messages"][-1]
        return ChatResponse(
            response=last_msg.content,
            thread_id=request.thread_id
        )
    except Exception as e:
        raise HTTPException(500, str(e))

@app.get("/threads/{thread_id}/history")
async def get_history(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = agent.get_state(config)
    return {"messages": [m.content for m in state.values["messages"]]}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Kubernetes

# k8s/langgraph-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langgraph-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langgraph-agent
  template:
    spec:
      containers:
      - name: agent
        image: your-registry/langgraph-agent:v1
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: openai-key
        - name: DATABASE_URL
          value: "postgresql://user:pass@postgres:5432/agent"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
---
apiVersion: v1
kind: Service
metadata:
  name: langgraph-agent
spec:
  selector:
    app: langgraph-agent
  ports:
  - port: 80
    targetPort: 8000

Testing Strategies

# test_agent.py
import pytest
from langgraph.graph import StateGraph

# Unit test nodes
@pytest.mark.asyncio
async def test_planner_node():
    state = {"research_topic": "AI agents", "messages": []}
    result = await planner(state)

    assert "sub_questions" in result
    assert len(result["sub_questions"]) >= 3
    assert result["current_step"] == "researching"

# Integration test with mocked LLM
@pytest.mark.asyncio
async def test_full_graph(mock_llm):
    agent = build_research_agent()

    config = {"configurable": {"thread_id": "test-1"}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content="Test topic")], "research_topic": "Test"},
        config
    )

    assert "final_report" in result
    assert len(result["final_report"]) > 100

# Property-based testing
from hypothesis import given, strategies as st

@given(st.text(min_size=10, max_size=200))
@pytest.mark.asyncio
async def test_agent_handles_any_input(topic):
    agent = build_research_agent()
    config = {"configurable": {"thread_id": f"test-{hash(topic)}"}}

    # Should not crash
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=topic)], "research_topic": topic},
        config
    )
    assert "messages" in result

Migration from LangChain Agents

# Old LangChain Agent
from langchain.agents import AgentExecutor, create_openai_tools_agent

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "Research AI trends"})

# New LangGraph (drop-in replacement)
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(llm, tools, checkpointer=MemorySaver())
result = agent.invoke(
    {"messages": [HumanMessage(content="Research AI trends")]},
    {"configurable": {"thread_id": "1"}}
)

Resources

• LangGraph Documentation
• LangGraph GitHub
• LangGraph Examples
• LangSmith Docs
• Agent Design Patterns
• Production Deployment Guide

---

LangGraph transforms agents from fragile prototypes into reliable, debuggable, production systems. Start simple, add persistence, then scale.