Building AI Agents with LangGraph: A Practical Guide

Create stateful, multi-step AI agents that can reason, use tools, and maintain context across complex workflows.

Introduction

LangGraph extends LangChain with graph-based orchestration — enabling cycles, branching, and persistent state. This is essential for building agents that:

• Handle multi-turn conversations with memory
• Execute complex workflows with decision points
• Recover from errors and retry
• Human-in-the-loop approval flows

Why LangGraph Over Chains?

Feature	LCEL Chains	LangGraph
Cycles/Loops	❌	✅
Branching	Limited	✅ Full control
State Persistence	Manual	✅ Built-in
Human-in-the-loop	Complex	✅ Native
Debugging	Hard	✅ Visual graph
Parallel Execution	❌	✅

Installation

pip install langgraph langchain-openai langchain-anthropic tavily-python

Core Concepts

1. State

The shared data structure that flows through the graph:

from typing import Annotated, List
from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import BaseMessage

class AgentState(MessagesState):
    # Extend MessagesState with custom fields
    user_id: str
    session_data: dict
    tool_call_count: int = 0

2. Nodes

Functions that process state and return updates:

from langchain_core.runnables import RunnableConfig

def agent_node(state: AgentState, config: RunnableConfig) -> AgentState:
    # Process messages, call LLM, return updated state
    return {"messages": [response]}

3. Edges

Control flow between nodes — conditional or unconditional:

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

Building a ReAct Agent with Tools

Complete Example

# agent.py
from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
import json

# 1. Define Tools
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    # In production, use Tavily, SerpAPI, etc.
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        # Safe eval for basic math
        allowed = set("0123456789+-*/(). ")
        if all(c in allowed for c in expression):
            return str(eval(expression))
        return "Invalid expression"
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Mock - replace with real API
    return f"Weather in {city}: Sunny, 22°C"

tools = [search_web, calculator, get_weather]

# 2. Define State
class AgentState(MessagesState):
    pass

# 3. Create LLM with Tools
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 4. Define Nodes
def agent(state: AgentState) -> AgentState:
    """Main agent reasoning node."""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

tool_node = ToolNode(tools)

# 5. Build Graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)

workflow.add_edge(START, "agent")

workflow.add_conditional_edges(
    "agent",
    lambda state: "tools" if state["messages"][-1].tool_calls else "end",
    {"tools": "tools", "end": END}
)

workflow.add_edge("tools", "agent")

app = workflow.compile()

# 6. Run
if __name__ == "__main__":
    # Initial state
    inputs = {
        "messages": [
            HumanMessage(content="What's 25 * 4 + 10? Also search for 'LangGraph tutorial'")
        ]
    }

    # Execute
    for chunk in app.stream(inputs):
        for node, output in chunk.items():
            print(f"\n--- {node} ---")
            if "messages" in output:
                msg = output["messages"][-1]
                print(f"{msg.type}: {msg.content}")
                if msg.tool_calls:
                    print(f"Tool calls: {msg.tool_calls}")

Adding Memory & Persistence

Short-term Memory (Thread-level)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# Run with thread_id for conversation memory
config = {"configurable": {"thread_id": "user-123"}}

# First message
result = app.invoke(
    {"messages": [HumanMessage(content="My name is Alice")]},
    config=config
)

# Second message - remembers name!
result = app.invoke(
    {"messages": [HumanMessage(content="What's my name?")]},
    config=config
)

Long-term Memory (Cross-session)

from langgraph.checkpoint.sqlite import SqliteSaver

# Persistent across restarts
conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

app = workflow.compile(checkpointer=checkpointer)

Human-in-the-Loop Approval

from langgraph.types import interrupt, Command

def sensitive_action(state: AgentState) -> AgentState:
    """Node that requires human approval."""
    # Pause execution and wait for human input
    decision = interrupt({
        "question": "Approve sending this email?",
        "draft": state["email_draft"]
    })

    if decision["approved"]:
        # Send email logic here
        return {"messages": [AIMessage(content="Email sent!")]}
    else:
        return {"messages": [AIMessage(content="Email cancelled.")]}

# Add to graph
workflow.add_node("human_approval", sensitive_action)
workflow.add_edge("agent", "human_approval")
workflow.add_edge("human_approval", "agent")

# Resume after human input
app.invoke(Command(resume={"approved": True}), config=config)

Multi-Agent Workflow

from langgraph.graph import StateGraph
from typing import Literal

class MultiAgentState(MessagesState):
    next_agent: str

# Specialist agents
def researcher(state: MultiAgentState) -> MultiAgentState:
    # Research logic
    return {"messages": [AIMessage(content="Research complete")], "next_agent": "writer"}

def writer(state: MultiAgentState) -> MultiAgentState:
    # Writing logic
    return {"messages": [AIMessage(content="Article written")], "next_agent": "editor"}

def editor(state: MultiAgentState) -> MultiAgentState:
    # Editing logic
    return {"messages": [AIMessage(content="Article edited")], "next_agent": "end"}

# Router
def router(state: MultiAgentState) -> Literal["researcher", "writer", "editor", "end"]:
    return state.get("next_agent", "end")

# Build
workflow = StateGraph(MultiAgentState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("editor", editor)

workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("researcher", router)
workflow.add_conditional_edges("writer", router)
workflow.add_conditional_edges("editor", router)

app = workflow.compile()

Streaming & Real-time Updates

# Stream tokens as they're generated
for chunk in app.stream(inputs, stream_mode="messages"):
    msg, metadata = chunk
    if msg.content:
        print(msg.content, end="", flush=True)

# Stream state updates
for chunk in app.stream(inputs, stream_mode="values"):
    print(f"State keys: {chunk.keys()}")

Debugging & Visualization

# Visualize graph
from langgraph.graph import draw_graph

# Save as PNG
draw_graph(app.get_graph()).write_png("agent_graph.png")

# Print Mermaid diagram
print(app.get_graph().draw_mermaid())

Output:

graph TD
    START --> agent
    agent -->|tools| tools
    agent -->|end| END
    tools --> agent

Production Checklist

• [ ] Error handling — Wrap nodes in try/catch, add retry logic
• [ ] Rate limiting — Use RunnableConfig for API limits
• [ ] Observability — Add LangSmith tracing
• [ ] Validation — Validate tool inputs/outputs with Pydantic
• [ ] Timeouts — Set execution timeouts per node
• [ ] State schema — Use Pydantic for strict state validation
• [ ] Testing — Unit test each node independently

# Example: Node with retry and timeout
from langgraph.graph import RetryPolicy

workflow.add_node(
    "unreliable_api",
    unreliable_node,
    retry=RetryPolicy(max_attempts=3, backoff_factor=2)
)

Advanced Patterns

1. Parallel Tool Execution

from langgraph.graph import StateGraph
from concurrent.futures import ThreadPoolExecutor

def parallel_tools(state: AgentState) -> AgentState:
    tool_calls = state["messages"][-1].tool_calls
    with ThreadPoolExecutor() as executor:
        results = list(executor.map(execute_tool, tool_calls))
    return {"messages": [ToolMessage(content=r, tool_call_id=tc["id"]) 
                            for r, tc in zip(results, tool_calls)]}

2. Dynamic Tool Selection

def select_tools(state: AgentState) -> list:
    """Choose tools based on query type."""
    query = state["messages"][-1].content.lower()
    if "calculate" in query:
        return [calculator]
    elif "weather" in query:
        return [get_weather]
    return [search_web]

3. Reflection & Self-Correction

def reflect(state: AgentState) -> AgentState:
    """Critique the last response and improve."""
    critique_prompt = f"Critique this answer: {state['messages'][-1].content}"
    critique = llm.invoke([HumanMessage(content=critique_prompt)])
    return {"messages": [critique]}

workflow.add_node("reflect", reflect)
workflow.add_edge("agent", "reflect")
workflow.add_conditional_edges("reflect", 
    lambda s: "agent" if "needs improvement" in s["messages"][-1].content.lower() else "end")

Deployment

FastAPI Server

# server.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    thread_id: str = "default"

@app.post("/chat")
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    result = agent_app.invoke(
        {"messages": [HumanMessage(content=request.message)]},
        config=config
    )
    return {"response": result["messages"][-1].content}

Docker

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

Resources

• LangGraph Docs
• LangGraph Examples
• LangSmith for Observability
• Agent Architectures

---

Part of the AI Agents series. Next: Multi-Agent Systems with LangGraph.