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

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