Advanced
Building a basic chatbot with an LLM is easy — give it a prompt, get a response. But real-world AI applications need more: they need to remember what happened three steps ago, decide which tool to call based on context, loop back when something goes wrong, and hand off to different processing nodes based on what the user needs. This is where LangGraph comes in. It is a Python library built on top of LangChain that lets you model AI workflows as stateful graphs — with nodes that do work, edges that route between them, and a shared state object that persists across every step.
LangGraph works with any LLM supported by LangChain, including OpenAI, Anthropic Claude, Google Gemini, and local models via Ollama. The core concepts — graphs, nodes, state, and conditional routing — are not tied to any specific model. You define the shape of your state, write Python functions for each node, connect them with edges, and LangGraph handles the execution loop. For local testing without an API key, we will show patterns that work with any chat model.
In this article we will cover how LangGraph works and when to use it over plain LangChain, how to define a TypedDict state schema, how to build nodes and edges, how to add conditional routing, how to integrate tool calling, and how to build a working multi-step agent. By the end you will have a complete LangGraph agent you can adapt for your own use case.
LangGraph in Python: Quick Example
Here is the smallest complete LangGraph program — a two-node graph that processes a message through a rewrite step and then a summary step, sharing state between them:
# quick_langgraph.py
from typing import TypedDict
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
message: str
rewritten: str
summary: str
def rewrite_node(state: AgentState) -> dict:
"""Capitalize and clean the input message."""
rewritten = state["message"].strip().capitalize() + "."
return {"rewritten": rewritten}
def summarize_node(state: AgentState) -> dict:
"""Produce a brief summary of the rewritten text."""
summary = f"Processed: {state['rewritten'][:50]}"
return {"summary": summary}
# Build the graph
graph = StateGraph(AgentState)
graph.add_node("rewrite", rewrite_node)
graph.add_node("summarize", summarize_node)
graph.set_entry_point("rewrite")
graph.add_edge("rewrite", "summarize")
graph.add_edge("summarize", END)
app = graph.compile()
result = app.invoke({"message": " hello world from langgraph ", "rewritten": "", "summary": ""})
print(result)
Output:
{'message': ' hello world from langgraph ', 'rewritten': 'Hello world from langgraph.', 'summary': 'Processed: Hello world from langgraph.'}
Each node is a plain Python function that receives the current state dict and returns a dict of updates to merge into the state. The graph wires the nodes together, and .invoke() runs the full pipeline. The state is automatically merged at each step — you only return the keys you changed, and LangGraph merges them with the existing state.
What Is LangGraph and When Should You Use It?
LangGraph models an AI workflow as a directed graph. Each node is a processing step (an LLM call, a tool invocation, a decision function). Each edge is a connection between steps. The state is a typed dictionary that flows through every node and accumulates results. Conditional edges let the graph branch based on state — routing to different nodes depending on what happened in the previous step.
The key difference between LangGraph and a simple function pipeline is that LangGraph supports cycles. A graph node can route back to an earlier node, enabling retry loops, multi-turn conversations, and agentic “think-act-observe” patterns where the agent can take multiple steps before returning a final answer.
| Scenario | Use LangChain chain | Use LangGraph |
|---|---|---|
| Simple prompt + response | Yes | Overkill |
| Multi-step pipeline (no loops) | Yes (LCEL) | Fine either way |
| Conditional branching | Hard | Yes |
| Retry loops / cycles | Not supported | Yes |
| Multi-agent coordination | Not supported | Yes |
| Persistent state across turns | Manual | Built-in with checkpointers |
Install LangGraph and the LangChain OpenAI integration with pip:
# terminal
pip install langgraph langchain langchain-openai
Defining the State Schema
Every LangGraph graph has a state schema defined as a TypedDict. This is the single source of truth for what data flows through the graph. Every node reads from this dict and returns a subset of it to update:
# state_schema.py
from typing import TypedDict, List, Optional
class ResearchState(TypedDict):
# Input
question: str
# Accumulated during the run
search_results: List[str]
analysis: str
# Control flow
iterations: int
max_iterations: int
# Final output
answer: Optional[str]
The TypedDict approach gives you type hints throughout your graph — your IDE can autocomplete state keys, and you get immediate feedback if a node returns an unexpected key. Some teams use Pydantic models instead for validation, but TypedDict is the standard pattern in LangGraph documentation and examples.
Building Nodes and Edges
Nodes are Python functions (sync or async) that take the state as input and return a dictionary of updates. Edges connect nodes and can be unconditional (always go to node B after node A) or conditional (go to B or C based on state):
# nodes_edges.py
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
class PipelineState(TypedDict):
text: str
tokens: List[str]
word_count: int
flagged: bool
def tokenize_node(state: PipelineState) -> dict:
tokens = state["text"].lower().split()
return {"tokens": tokens, "word_count": len(tokens)}
def check_content_node(state: PipelineState) -> dict:
bad_words = {"spam", "scam", "free", "click"}
flagged = any(t in bad_words for t in state["tokens"])
return {"flagged": flagged}
def approve_node(state: PipelineState) -> dict:
print(f"APPROVED: '{state['text']}' ({state['word_count']} words)")
return {}
def reject_node(state: PipelineState) -> dict:
print(f"REJECTED: '{state['text']}' -- contains flagged content")
return {}
def route_by_flag(state: PipelineState) -> str:
"""Conditional edge: route to approve or reject based on flagged field."""
return "reject" if state["flagged"] else "approve"
graph = StateGraph(PipelineState)
graph.add_node("tokenize", tokenize_node)
graph.add_node("check_content", check_content_node)
graph.add_node("approve", approve_node)
graph.add_node("reject", reject_node)
graph.set_entry_point("tokenize")
graph.add_edge("tokenize", "check_content")
graph.add_conditional_edges("check_content", route_by_flag, {
"approve": "approve",
"reject": "reject",
})
graph.add_edge("approve", END)
graph.add_edge("reject", END)
app = graph.compile()
app.invoke({"text": "Buy our product today", "tokens": [], "word_count": 0, "flagged": False})
app.invoke({"text": "Click here for free spam", "tokens": [], "word_count": 0, "flagged": False})
Output:
APPROVED: 'Buy our product today' (4 words)
REJECTED: 'Click here for free spam' -- contains flagged content
The add_conditional_edges call takes the source node, a routing function that returns a string, and a mapping from those strings to destination nodes. This is how LangGraph implements branching — the routing function inspects the state and returns a key that maps to the next node to run.
Adding Tool Calling
LangGraph agents commonly use tools — Python functions the LLM can invoke. Here is a self-contained example that defines tools and routes through them without requiring an API key, using a mock LLM call:
# tool_calling.py
import json
from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, END
# --- Define tools ---
def get_weather(city: str) -> str:
"""Mock weather tool."""
data = {
"Sydney": "28C, sunny",
"London": "12C, cloudy",
"New York": "18C, partly cloudy",
}
return data.get(city, "Weather data unavailable")
def calculate(expression: str) -> str:
"""Safe math evaluator."""
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"Error: {e}"
TOOLS = {"get_weather": get_weather, "calculate": calculate}
# --- Graph state ---
class ToolState(TypedDict):
request: Dict[str, Any] # {"tool": "...", "args": {...}}
tool_result: str
final_answer: str
def call_tool_node(state: ToolState) -> dict:
req = state["request"]
tool_fn = TOOLS.get(req["tool"])
if tool_fn:
result = tool_fn(**req["args"])
else:
result = f"Unknown tool: {req['tool']}"
return {"tool_result": result}
def format_answer_node(state: ToolState) -> dict:
answer = f"Tool '{state['request']['tool']}' returned: {state['tool_result']}"
return {"final_answer": answer}
graph = StateGraph(ToolState)
graph.add_node("call_tool", call_tool_node)
graph.add_node("format", format_answer_node)
graph.set_entry_point("call_tool")
graph.add_edge("call_tool", "format")
graph.add_edge("format", END)
app = graph.compile()
# Test weather tool
r1 = app.invoke({"request": {"tool": "get_weather", "args": {"city": "Sydney"}},
"tool_result": "", "final_answer": ""})
print(r1["final_answer"])
# Test calculator tool
r2 = app.invoke({"request": {"tool": "calculate", "args": {"expression": "250 * 1.1"}},
"tool_result": "", "final_answer": ""})
print(r2["final_answer"])
Output:
Tool 'get_weather' returned: 28C, sunny
Tool 'calculate' returned: 275.0
In production, the tool-calling node would parse the LLM output, extract the tool name and arguments from the model’s JSON response, call the tool, and return the result to feed back into the LLM in the next iteration. LangChain provides ToolNode and tools_condition helpers that automate this pattern for OpenAI function-calling and Anthropic tool-use formats.
Real-Life Example: Multi-Step Research Agent
This example builds a complete research agent that breaks a question into sub-tasks, processes each one, checks quality, and loops if needed — demonstrating cycles, state accumulation, and conditional routing in a realistic pattern:
# research_agent.py
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
class ResearchState(TypedDict):
question: str
sub_questions: List[str]
answers: List[str]
current_index: int
quality_score: int
final_report: str
def decompose_node(state: ResearchState) -> dict:
"""Break the main question into sub-questions."""
q = state["question"]
sub_qs = [
f"What is the definition of: {q}?",
f"What are the main use cases of: {q}?",
f"What are the limitations of: {q}?",
]
print(f"Decomposed into {len(sub_qs)} sub-questions.")
return {"sub_questions": sub_qs, "current_index": 0}
def research_node(state: ResearchState) -> dict:
"""Process the current sub-question (mock LLM call)."""
idx = state["current_index"]
sq = state["sub_questions"][idx]
# In production: call your LLM here
answer = f"[Answer to '{sq}' -- expand with LLM call]"
answers = state["answers"] + [answer]
print(f"Researched sub-question {idx + 1}/{len(state['sub_questions'])}")
return {"answers": answers, "current_index": idx + 1}
def should_continue(state: ResearchState) -> str:
"""Continue researching or move to synthesis."""
if state["current_index"] < len(state["sub_questions"]):
return "continue"
return "synthesize"
def synthesize_node(state: ResearchState) -> dict:
"""Combine answers into a final report."""
report = f"Research Report: {state['question']}\n"
report += "\n".join(f" - {a}" for a in state["answers"])
score = min(len(state["answers"]) * 30, 90)
return {"final_report": report, "quality_score": score}
def quality_check(state: ResearchState) -> str:
return "done" if state["quality_score"] >= 70 else "redo"
def finish_node(state: ResearchState) -> dict:
print("\n=== FINAL REPORT ===")
print(state["final_report"])
return {}
graph = StateGraph(ResearchState)
graph.add_node("decompose", decompose_node)
graph.add_node("research", research_node)
graph.add_node("synthesize", synthesize_node)
graph.add_node("finish", finish_node)
graph.set_entry_point("decompose")
graph.add_edge("decompose", "research")
graph.add_conditional_edges("research", should_continue, {
"continue": "research",
"synthesize": "synthesize",
})
graph.add_conditional_edges("synthesize", quality_check, {
"done": "finish",
"redo": "decompose",
})
graph.add_edge("finish", END)
app = graph.compile()
result = app.invoke({
"question": "Python asyncio",
"sub_questions": [], "answers": [],
"current_index": 0, "quality_score": 0, "final_report": ""
})
Output:
Decomposed into 3 sub-questions.
Researched sub-question 1/3
Researched sub-question 2/3
Researched sub-question 3/3
=== FINAL REPORT ===
Research Report: Python asyncio
- [Answer to 'What is the definition of: Python asyncio?' ...]
- [Answer to 'What are the main use cases of: Python asyncio?' ...]
- [Answer to 'What are the limitations of: Python asyncio?' ...]
The research node cycles back to itself via the should_continue conditional edge — this is the loop that processes one sub-question per iteration. Once all sub-questions are answered, it routes to synthesize. The quality check then either finishes or loops back to decompose to try again. Replace the mock LLM calls with ChatOpenAI or ChatAnthropic to make this a real research agent.
Frequently Asked Questions
Does LangGraph support streaming?
Yes. Use app.stream(initial_state) instead of app.invoke(). This returns an iterator that yields the state update after each node completes, letting you stream intermediate results to a UI or log in real time. You can also use app.astream() for async streaming with async for. Streaming is particularly useful when one node’s output needs to be displayed while the next node is still processing.
How do I add persistent memory to a LangGraph agent?
LangGraph supports checkpointers that save state between invocations. Use from langgraph.checkpoint.sqlite import SqliteSaver for local persistence or from langgraph.checkpoint.postgres import PostgresSaver for production. Pass the checkpointer to graph.compile(checkpointer=saver). Then provide a thread_id in the config when invoking: app.invoke(state, config={"configurable": {"thread_id": "user-123"}}). This enables multi-turn conversations where the agent remembers previous exchanges.
Can LangGraph handle multiple agents?
Yes. LangGraph supports supervisor patterns where one agent node routes tasks to specialized sub-agent nodes. Each sub-agent can itself be a compiled LangGraph graph, and you can use app.invoke() inside a node to call a sub-agent. This is the recommended pattern for building complex systems like a coding agent that delegates browser searches to a web-search specialist agent and code execution to a sandbox agent.
How does LangGraph compare to AutoGen or CrewAI?
LangGraph gives you explicit control over the graph structure, state, and routing logic — you define exactly how agents communicate. AutoGen and CrewAI abstract more away and are easier to start with but harder to customize. LangGraph is the better choice when you need precise control over agent behavior, deterministic routing, or integration with an existing LangChain codebase. AutoGen and CrewAI are better for quickly prototyping multi-agent conversations without writing graph code.
How do I debug a LangGraph graph?
LangGraph Studio (the desktop GUI) is the best debugging tool — it visualizes the graph, lets you step through nodes, and shows state at each step. For terminal debugging, use app.stream() with a print statement after each yielded event: for event in app.stream(state): print(event). You can also enable LangSmith tracing by setting LANGCHAIN_TRACING_V2=true in your environment, which logs every node invocation with inputs and outputs to the LangSmith cloud dashboard.
Conclusion
LangGraph makes stateful multi-step AI agents practical to build in Python. We covered the core building blocks — state schemas with TypedDict, nodes as plain Python functions, unconditional and conditional edges for routing, cycles for loops, and tool-calling patterns. The research agent example showed how these pieces combine into an agent that decomposes problems, processes them iteratively, and checks its own output quality before finishing.
The natural next step is to replace the mock processing in the research agent with real LLM calls using ChatOpenAI or ChatAnthropic, then add a SqliteSaver checkpointer to give it persistent memory. The official LangGraph documentation has excellent tutorials on both patterns, including pre-built ReAct agent templates that handle tool-calling boilerplate for you.