LangGraph¶
Overview¶
LangGraph is a framework for building agent workflows as directed graphs. Perfect for complex, multi-step agent systems with loops and conditional logic.
StateGraph Basics¶
from langgraph.graph import StateGraph
from typing import TypedDict
class AgentState(TypedDict):
"""Shared agent state"""
messages: list
next: str
final_answer: str = None
class ResearchAgent:
"""Build agent with LangGraph"""
def __init__(self):
self.workflow = StateGraph(AgentState)
self.setup_workflow()
def setup_workflow(self):
"""Define graph structure"""
# Add nodes
self.workflow.add_node("research", self.research_node)
self.workflow.add_node("analyze", self.analyze_node)
self.workflow.add_node("synthesize", self.synthesize_node)
# Add edges
self.workflow.add_edge("research", "analyze")
self.workflow.add_edge("analyze", "synthesize")
# Set entry point
self.workflow.set_entry_point("research")
# Compile
self.graph = self.workflow.compile()
def research_node(self, state: AgentState):
"""Research phase"""
query = state["messages"][-1]
results = self.search(query)
return {
"messages": state["messages"] + [f"Found: {results}"],
"next": "analyze"
}
def analyze_node(self, state: AgentState):
"""Analysis phase"""
findings = state["messages"][-1]
analysis = self.analyze(findings)
return {
"messages": state["messages"] + [f"Analysis: {analysis}"],
"next": "synthesize"
}
def synthesize_node(self, state: AgentState):
"""Final synthesis"""
analysis = state["messages"][-1]
final = self.synthesize(analysis)
return {
"messages": state["messages"],
"final_answer": final
}
def run(self, query: str):
"""Execute workflow"""
initial_state = {
"messages": [query],
"next": "research"
}
result = self.graph.invoke(initial_state)
return result["final_answer"]
Conditional Routing¶
Dynamic Decision Making¶
class ConditionalWorkflow:
"""Route based on conditions"""
def __init__(self):
self.workflow = StateGraph(AgentState)
self.setup_conditional_routing()
def setup_conditional_routing(self):
"""Add conditional edges"""
self.workflow.add_node("classify", self.classify_node)
self.workflow.add_node("simple_response", self.simple_node)
self.workflow.add_node("complex_analysis", self.complex_node)
# Conditional routing
self.workflow.add_conditional_edges(
"classify",
self.route_based_on_complexity, # Routing function
{
"simple": "simple_response",
"complex": "complex_analysis"
}
)
self.workflow.set_entry_point("classify")
self.graph = self.workflow.compile()
def classify_node(self, state: AgentState):
"""Classify query difficulty"""
query = state["messages"][-1]
complexity = self.estimate_complexity(query)
return {
"messages": state["messages"],
"complexity": complexity
}
def route_based_on_complexity(self, state: AgentState):
"""Route based on complexity"""
if state["complexity"] < 0.5:
return "simple"
else:
return "complex"
def simple_node(self, state):
"""Handle simple queries"""
answer = self.quick_answer(state["messages"][-1])
return {"messages": state["messages"], "final": answer}
def complex_node(self, state):
"""Handle complex queries"""
answer = self.deep_analysis(state["messages"][-1])
return {"messages": state["messages"], "final": answer}
Loops & Reflection¶
Self-Improvement Cycles¶
class ReflectiveWorkflow:
"""Agent that reflects on its work"""
def __init__(self):
self.workflow = StateGraph(AgentState)
self.setup_reflection_loop()
def setup_reflection_loop(self):
"""Create loop with reflection"""
self.workflow.add_node("generate", self.generate_node)
self.workflow.add_node("critique", self.critique_node)
self.workflow.add_node("revise", self.revise_node)
# Create loop
self.workflow.add_edge("generate", "critique")
# Conditional: should we revise?
self.workflow.add_conditional_edges(
"critique",
self.should_revise,
{
"yes": "revise",
"no": "end"
}
)
self.workflow.add_edge("revise", "generate") # Loop back
self.workflow.set_entry_point("generate")
self.graph = self.workflow.compile()
def generate_node(self, state):
"""Generate solution"""
solution = self.generate_solution(state["messages"][-1])
return {"messages": state["messages"] + [solution]}
def critique_node(self, state):
"""Critique solution"""
critique = self.critique(state["messages"][-1])
return {
"messages": state["messages"] + [critique],
"critique_score": self.score_critique(critique)
}
def should_revise(self, state):
"""Decide if we should revise"""
score = state["critique_score"]
return "yes" if score < 0.7 else "no"
def revise_node(self, state):
"""Improve solution based on critique"""
revised = self.improve(state["messages"])
return {"messages": state["messages"] + [revised]}
Streaming & Monitoring¶
Real-Time Output¶
class MonitoredWorkflow:
"""Monitor workflow execution"""
def run_with_streaming(self, query):
"""Execute with streaming"""
initial_state = {"messages": [query], "next": None}
# Stream events
for event in self.graph.stream(initial_state):
node_name = list(event.keys())[0]
node_output = event[node_name]
print(f"Node: {node_name}")
print(f"Output: {node_output}")
yield node_output
3 Warnings¶
Warning 1: Infinite Loops¶
# WRONG
# Loop without termination condition
workflow.add_edge("revise", "generate") # Always loops
# RIGHT
# Use conditional edge to exit
workflow.add_conditional_edges(
"revise",
should_continue,
{
"continue": "generate",
"end": END
}
)
Warning 2: State Explosion¶
# WRONG
# State grows unbounded
state["messages"].append(every_output)
# Memory usage explodes
# RIGHT
# Clean up old state
if len(state["messages"]) > 20:
state["messages"] = state["messages"][-20:]
Warning 3: Overcomplex Graphs¶
# WRONG
# Graph with too many nodes
# Hard to understand and debug
# RIGHT
# Start simple, add complexity if needed
# 3-5 nodes initially
# Add more only when necessary
-
Last Updated: August 9, 2026