LangChain is excellent for linear pipelines. Prompt goes in, chain executes left to right, answer comes out. Clean and predictable.
But real AI agents don't work that way. They need to:
- Loop: try something, check if it worked, retry if it didn't
- Branch: take different paths based on LLM output
- Persist state: share a scratchpad across multiple steps
- Use tools: call APIs, search the web, run code
LangGraph was built specifically for this. It models your agent as a directed graph — nodes are functions or LLM calls, edges define the flow between them, and a shared state object ties everything together.
This guide covers LangGraph from scratch with working code examples.
Installation
pip install langchain langgraph langchain-openai
Core Concepts
State
State is a TypedDict that every node reads from and writes to. Think of it as the agent's shared memory for one run.
from typing import TypedDict, List, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[List[str], operator.add] # accumulates across nodes
current_step: str
final_answer: str
Annotated[List[str], operator.add] means each node appends to the list rather than replacing it — a common pattern for message history.
Nodes
A node is any Python function that takes state and returns a dict of updates:
def my_node(state: AgentState) -> dict:
# read from state
last_message = state["messages"][-1]
# do something (LLM call, tool call, logic)
result = do_work(last_message)
# return only the keys you're updating
return {"current_step": "done", "final_answer": result}
Edges
Edges connect nodes. They can be static (always go from A to B) or conditional (go to A or B based on state).
Your First LangGraph Agent
A simple two-node graph: one node calls the LLM, one node formats the output.
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, List
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class State(TypedDict):
messages: List
response: str
def call_llm(state: State) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response], "response": response.content}
def format_output(state: State) -> dict:
# Could add post-processing, logging, etc.
formatted = f"Answer: {state['response']}"
return {"response": formatted}
# Build the graph
graph = StateGraph(State)
graph.add_node("call_llm", call_llm)
graph.add_node("format_output", format_output)
graph.set_entry_point("call_llm")
graph.add_edge("call_llm", "format_output")
graph.add_edge("format_output", END)
app = graph.compile()
# Run it
result = app.invoke({
"messages": [HumanMessage(content="What is chain of thought prompting?")],
"response": ""
})
print(result["response"])
Conditional Edges — Branching Logic
This is where LangGraph gets powerful. Route to different nodes based on the current state.
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class State(TypedDict):
question: str
category: str
answer: str
def classify_question(state: State) -> dict:
"""Classify the question as 'factual', 'creative', or 'code'."""
response = llm.invoke(
f"Classify this question as exactly one of: factual, creative, code.\n"
f"Question: {state['question']}\nReturn only the category word."
)
return {"category": response.content.strip().lower()}
def answer_factual(state: State) -> dict:
response = llm.invoke(
f"Answer this factual question precisely: {state['question']}"
)
return {"answer": response.content}
def answer_creative(state: State) -> dict:
response = llm.invoke(
f"Answer this creatively and with detail: {state['question']}"
)
return {"answer": response.content}
def answer_code(state: State) -> dict:
response = llm.invoke(
f"Provide working code to answer: {state['question']}. Include comments."
)
return {"answer": response.content}
def route_question(state: State) -> Literal["factual", "creative", "code"]:
"""Return the name of the next node to visit."""
return state["category"]
# Build
graph = StateGraph(State)
graph.add_node("classify", classify_question)
graph.add_node("factual", answer_factual)
graph.add_node("creative", answer_creative)
graph.add_node("code", answer_code)
graph.set_entry_point("classify")
# Conditional edge — inspects state after 'classify' to choose next node
graph.add_conditional_edges(
"classify",
route_question,
{"factual": "factual", "creative": "creative", "code": "code"}
)
graph.add_edge("factual", END)
graph.add_edge("creative", END)
graph.add_edge("code", END)
app = graph.compile()
result = app.invoke({"question": "Write a Python function to reverse a string", "category": "", "answer": ""})
print(result["answer"])
Cycles — The Self-Correcting Agent
The key advantage of LangGraph over plain chains: you can loop. Here's an agent that keeps retrying until the output meets a quality check.
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
import json
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
class State(TypedDict):
task: str
draft: str
feedback: str
iterations: int
max_iterations: int
def write_draft(state: State) -> dict:
prompt = state["task"]
if state["draft"]:
prompt = (
f"Rewrite the following draft based on this feedback: {state['feedback']}\n\n"
f"Original draft: {state['draft']}\n\n"
f"Task: {state['task']}"
)
response = llm.invoke(prompt)
return {"draft": response.content, "iterations": state["iterations"] + 1}
def critique_draft(state: State) -> dict:
"""Grade the draft and provide improvement feedback."""
response = llm.invoke(
f"Evaluate this draft for the task: '{state['task']}'\n\n"
f"Draft: {state['draft']}\n\n"
f"Return JSON: {{\"score\": 1-10, \"feedback\": \"specific improvement notes\", \"approved\": true/false}}"
)
try:
evaluation = json.loads(response.content)
except json.JSONDecodeError:
evaluation = {"score": 5, "feedback": "Needs improvement", "approved": False}
return {"feedback": evaluation.get("feedback", ""), **evaluation}
def should_continue(state: State) -> Literal["write", "end"]:
"""Loop back to write_draft if not approved, stop if approved or max iterations reached."""
evaluation = json.loads(state.get("feedback", "{}")) if isinstance(state.get("feedback"), str) else {}
approved = state.get("approved", False)
if approved or state["iterations"] >= state["max_iterations"]:
return "end"
return "write"
graph = StateGraph(State)
graph.add_node("write", write_draft)
graph.add_node("critique", critique_draft)
graph.set_entry_point("write")
graph.add_edge("write", "critique")
graph.add_conditional_edges("critique", should_continue, {"write": "write", "end": END})
app = graph.compile()
result = app.invoke({
"task": "Write a 3-sentence product description for an AI-powered writing assistant",
"draft": "",
"feedback": "",
"iterations": 0,
"max_iterations": 3,
})
print(f"Final draft (after {result['iterations']} iterations):")
print(result["draft"])
Tool-Using Agents
The most common LangGraph pattern: an agent that can call tools, observe results, and decide what to do next.
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from typing import TypedDict, List, Annotated
import operator
# Define tools
@tool
def search_web(query: str) -> str:
"""Search the web for current information."""
# In production: integrate with Tavily, SerpAPI, etc.
return f"[Search results for '{query}': Found relevant information about {query}]"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
tools = [search_web, calculate]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
def agent_node(state: AgentState) -> dict:
"""Call the LLM. It may return a tool call or a final answer."""
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
def should_use_tool(state: AgentState) -> str:
"""Check if the last message contains a tool call."""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"
# ToolNode automatically executes whatever tool the LLM requested
tool_node = ToolNode(tools)
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_use_tool, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent") # after tool runs, go back to agent
app = graph.compile()
result = app.invoke({
"messages": [HumanMessage(content="What is 1234 * 5678? Show your work.")]
})
# Print the conversation
for message in result["messages"]:
role = message.__class__.__name__
print(f"{role}: {message.content[:200]}")
Checkpointing — Persist Agent State
In production you need to save agent state between runs (for resuming, for user sessions). LangGraph supports this via checkpointers.
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver # for production
# In-memory (good for development)
memory = MemorySaver()
# SQLite (persists to disk)
# memory = SqliteSaver.from_conn_string("./agent_state.db")
app = graph.compile(checkpointer=memory)
# Each thread_id is an independent conversation
config = {"configurable": {"thread_id": "user_456_session_1"}}
# Turn 1
result1 = app.invoke(
{"messages": [HumanMessage(content="My name is Arjun.")]},
config=config
)
# Turn 2 — state is preserved, agent remembers "Arjun"
result2 = app.invoke(
{"messages": [HumanMessage(content="What's my name?")]},
config=config
)
print(result2["messages"][-1].content) # "Your name is Arjun."
Inspecting Agent Runs
For debugging, stream the graph execution step by step:
for event in app.stream(
{"messages": [HumanMessage(content="Search for the latest on LLM benchmarks")]},
config={"configurable": {"thread_id": "debug_run_1"}}
):
for node_name, output in event.items():
print(f"--- Node: {node_name} ---")
if "messages" in output:
print(output["messages"][-1].content[:300])
LangGraph vs Plain LangChain
| Need | Use |
|---|---|
| Linear: A → B → C | LangChain LCEL |
| Branching: go to A or B | LangGraph conditional edges |
| Looping: retry until good | LangGraph cycles |
| Tool use with multi-step reasoning | LangGraph + ToolNode |
| Persistent state across turns | LangGraph checkpointer |
| Human-in-the-loop (pause for approval) | LangGraph interrupt_before |
The Pattern to Remember
Every LangGraph agent follows the same structure:
- Define state — what information persists across the whole run
- Define nodes — each does one thing (LLM call, tool call, validation)
- Define edges — how nodes connect, what conditions trigger which path
- Compile and run —
graph.compile()thenapp.invoke()
The power is in the cycles and conditional edges. Once you have those, you can build agents that retry on failure, route to specialists, coordinate multiple sub-agents, and maintain long-running conversations — all with clean, testable code.
For the full picture on when to reach for LangGraph vs plain chains, see LangChain vs LangGraph — When to Use Each.