LangGraph: Advanced Patterns π§©
The Story of the Master Architect
Imagine youβre building the worldβs most amazing theme park. You started with simple rides, but now you want:
- Rides that talk to each other π’
- Workers who ask visitors for permission before doing scary things ποΈ
- Multiple teams working together on huge attractions π·ββοΈπ·ββοΈ
- Emergency systems when things go wrong π¨
- A control room to watch everything πΊ
Thatβs exactly what Advanced LangGraph Patterns give you for AI apps!
1. LangGraph Functional API π§
The Simple Way to Build
Think of the Functional API like LEGO instructions β step-by-step, easy to follow.
Instead of creating complex classes, you just write simple functions and connect them.
Before (Class-based):
class MyNode:
def __call__(self, state):
return {"data": "result"}
After (Functional API):
@task
def my_task(input):
return "result"
@entrypoint()
def my_graph(input):
result = my_task(input)
return result
Key Parts
| Part | What It Does | Like⦠|
|---|---|---|
@task |
Marks a function as a step | A single LEGO piece |
@entrypoint() |
The starting point | The first instruction |
interrupt() |
Pause and wait | βStop! Ask mom first!β |
Why Use It?
- β Simpler code β less typing
- β Easier to read β looks like normal Python
- β Same power β does everything the old way could
graph TD A["Start: entrypoint"] --> B["Task 1"] B --> C["Task 2"] C --> D["Return Result"]
2. Subgraphs ποΈ
Graphs Inside Graphs!
Remember our theme park? A subgraph is like having a mini-park inside the main park.
The water park area has its own rides, but itβs still part of the big park!
Real Example
# Mini-graph for research
research_graph = StateGraph(...)
research_graph.add_node("search", search_web)
research_graph.add_node("summarize", summarize)
# Main graph uses the mini-graph
main_graph = StateGraph(...)
main_graph.add_node("research",
research_graph.compile())
main_graph.add_node("write", write_report)
When to Use Subgraphs
| Situation | Use Subgraph? |
|---|---|
| Same steps used in many places | β Yes! |
| Complex task with many parts | β Yes! |
| Simple one-time step | β No need |
The Magic Flow
graph TD A["Main Graph Start"] --> B["Subgraph"] B --> C["Sub-step 1"] C --> D["Sub-step 2"] D --> E["Back to Main"] E --> F["Continue Main"]
Key Insight: The main graph doesnβt know (or care) what happens inside the subgraph. It just waits for the result!
3. Prebuilt ReAct Agent π€
The Ready-Made Robot Helper
Building an AI agent from scratch is like building a car from raw metal. But what if someone already built the car for you?
ReAct = Reason + Act
The agent thinks, then does, then thinks again!
Using the Prebuilt Agent
from langgraph.prebuilt import create_react_agent
# Your tools (things the agent can do)
tools = [search_tool, calculator_tool]
# Create agent in ONE line!
agent = create_react_agent(
model=my_llm,
tools=tools
)
# Run it
result = agent.invoke({
"messages": [
("user", "What is 25 * 4?")
]
})
What Happens Inside
graph TD A["Get Question"] --> B{Need a Tool?} B -->|Yes| C["Pick Tool"] C --> D["Use Tool"] D --> E["Get Result"] E --> B B -->|No| F["Give Final Answer"]
Built-in Superpowers
| Feature | What You Get |
|---|---|
| Tool calling | Agent picks the right tool |
| Memory | Remembers conversation |
| Retries | Tries again if tool fails |
| Streaming | See answers as they come |
4. Human-in-the-Loop Patterns π
Asking Permission First
Some AI decisions are too important to make alone. Like asking a parent before buying something expensive!
The Three Patterns
Pattern 1: Approve Before Action
from langgraph.types import interrupt
def risky_action(state):
# Pause and ask human
human_says = interrupt(
"Should I delete this file?"
)
if human_says == "yes":
delete_file()
return state
Pattern 2: Edit the Plan
def make_plan(state):
plan = ai_makes_plan()
# Human can change it
edited_plan = interrupt(plan)
return {"plan": edited_plan}
Pattern 3: Review Output
def generate_email(state):
email = ai_writes_email()
# Human approves or edits
final = interrupt(email)
return {"email": final}
Visual Flow
graph TD A["AI Does Work"] --> B["PAUSE"] B --> C{Human Reviews} C -->|Approve| D["Continue"] C -->|Edit| E["AI Gets Edits"] E --> D C -->|Reject| F["Try Again"] F --> A
The Command Object
from langgraph.types import Command
# Resume with approval
graph.invoke(
Command(resume="approved"),
config=thread_config
)
5. Multi-Agent Systems π€
A Team of AI Helpers
One robot is good. But a team of specialized robots? Unstoppable!
Types of Multi-Agent Setups
Type A: Supervisor Pattern
One boss agent tells others what to do.
def supervisor(state):
# Decide who works next
if need_research:
return "researcher"
elif need_writing:
return "writer"
else:
return "end"
graph TD S["Supervisor"] --> R["Researcher"] S --> W["Writer"] S --> E["Editor"] R --> S W --> S E --> S
Type B: Network Pattern
Agents talk to each other freely.
def researcher(state):
findings = do_research()
# Pass directly to writer
return Command(
goto="writer",
update={"research": findings}
)
Type C: Hierarchical Pattern
Teams with their own supervisors!
graph TD CEO["Top Supervisor"] --> TL1["Team Lead 1"] CEO --> TL2["Team Lead 2"] TL1 --> W1["Worker 1"] TL1 --> W2["Worker 2"] TL2 --> W3["Worker 3"] TL2 --> W4["Worker 4"]
Sharing Information
# Shared state all agents can see
class SharedState(TypedDict):
messages: list
current_task: str
results: dict
6. Error Handling in Graphs π¨
When Things Go Wrong
Even the best theme parks have problems. The key is handling them gracefully!
Strategy 1: Try-Catch in Nodes
def risky_node(state):
try:
result = call_external_api()
return {"data": result}
except APIError:
return {"error": "API failed"}
Strategy 2: Retry Logic
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def api_node(state):
return call_flaky_api()
Strategy 3: Fallback Nodes
def router(state):
if state.get("error"):
return "fallback_node"
return "normal_node"
Strategy 4: Error State
class State(TypedDict):
data: str
error: Optional[str]
retry_count: int
Error Flow
graph TD A["Try Action"] --> B{Success?} B -->|Yes| C["Continue"] B -->|No| D{Retries Left?} D -->|Yes| A D -->|No| E["Fallback"] E --> F["Graceful End"]
Best Practices
| Do This | Not This |
|---|---|
| Log errors clearly | Silently fail |
| Have fallback plans | Crash the whole graph |
| Limit retries (3-5) | Retry forever |
| Tell the user | Hide problems |
7. LangGraph Studio Debugging π
Your Theme Park Control Room
LangGraph Studio is like having cameras and controls for every ride in your park!
What You Can See
- π Graph visualization β See all nodes and edges
- π Live execution β Watch data flow in real-time
- βΈοΈ Breakpoints β Pause at any node
- π State inspection β See whatβs in state
- βͺ Time travel β Go back to any step
Starting Studio
# Install
pip install langgraph-cli
# Run in your project folder
langgraph dev
The Interface
βββββββββββββββββββββββββββββββββββββββ
β π GRAPH VIEW β
β [Node A] β [Node B] β [Node C] β
βββββββββββββββββββββββββββββββββββββββ€
β π STATE β
β messages: [...] β
β current_node: "Node B" β
βββββββββββββββββββββββββββββββββββββββ€
β βΆοΈ CONTROLS β
β [Step] [Continue] [Reset] β
βββββββββββββββββββββββββββββββββββββββ
Debugging Tips
Tip 1: Use Thread IDs
config = {"configurable": {"thread_id": "debug-123"}}
result = graph.invoke(input, config)
Tip 2: Add Checkpoints
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(
checkpointer=MemorySaver()
)
Tip 3: Inspect State
# Get state at any point
state = graph.get_state(config)
print(state.values)
print(state.next) # What runs next?
Common Bugs & Fixes
| Bug | Fix |
|---|---|
| Infinite loop | Check your routing conditions |
| Missing data | Verify state updates |
| Wrong node runs | Check edge definitions |
| State not saved | Add a checkpointer |
graph TD A["Bug Found!"] --> B["Open Studio"] B --> C["Find the Node"] C --> D["Check State"] D --> E["Fix Code"] E --> F["Test Again"]
Putting It All Together π―
Hereβs how all these patterns work in a real app:
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
# 1. Create specialized agents
researcher = create_react_agent(model, research_tools)
writer = create_react_agent(model, writing_tools)
# 2. Build main graph with subgraphs
builder = StateGraph(State)
builder.add_node("research", researcher)
builder.add_node("write", writer)
builder.add_node("review", human_review) # Human-in-loop
# 3. Add error handling
builder.add_node("error_handler", handle_errors)
# 4. Connect with routing
builder.add_conditional_edges(
"research",
route_after_research
)
# 5. Compile with checkpointing for Studio
graph = builder.compile(
checkpointer=MemorySaver()
)
Quick Reference Card π
| Pattern | Use When⦠|
|---|---|
| Functional API | You want simpler code |
| Subgraphs | Reusing or organizing logic |
| ReAct Agent | Need quick tool-using agent |
| Human-in-Loop | Critical decisions need approval |
| Multi-Agent | Complex tasks need specialists |
| Error Handling | Things might fail |
| Studio | Debugging and monitoring |
You Did It! π
You now understand the advanced patterns that power production LangGraph apps:
- β Functional API β Simple, clean code
- β Subgraphs β Organize complex flows
- β ReAct Agent β Ready-made intelligence
- β Human-in-Loop β Safe, supervised AI
- β Multi-Agent β Teams of AI helpers
- β Error Handling β Graceful failures
- β Studio β See everything, fix anything
Youβre ready to build amazing AI applications! π
