LangGraph: Stateful Agent Workflows¶
Building stateful, multi-step agent workflows with LangGraph — nodes, edges, state management, and human-in-the-loop.
# Install dependencies
# !pip install langgraph langchain-openai
Basic LangGraph Agent¶
from typing import TypedDict, Annotated, List
import operator
# Define state
class AgentState(TypedDict):
"""State of the agent"""
query: str
thoughts: Annotated[List[str], operator.add] # Accumulate thoughts
observations: Annotated[List[str], operator.add] # Accumulate observations
answer: str
iteration: int
# Example state
state = AgentState(
query="What's the capital of France?",
thoughts=[],
observations=[],
answer="",
iteration=0
)
print("Agent State Structure:")
print(state)
# Simplified LangGraph implementation (no external dependencies)
from typing import Callable, Dict, List, Optional
class SimpleGraph:
"""Simplified graph execution engine"""
def __init__(self):
self.nodes: Dict[str, Callable] = {}
self.edges: Dict[str, str] = {}
self.conditional_edges: Dict[str, Callable] = {}
self.entry_point = None
def add_node(self, name: str, func: Callable):
"""Add a node to the graph"""
self.nodes[name] = func
def add_edge(self, from_node: str, to_node: str):
"""Add a direct edge"""
self.edges[from_node] = to_node
def add_conditional_edge(self, from_node: str, condition: Callable):
"""Add a conditional edge"""
self.conditional_edges[from_node] = condition
def set_entry_point(self, node: str):
"""Set the starting node"""
self.entry_point = node
def run(self, initial_state: Dict) -> Dict:
"""Execute the graph"""
state = initial_state.copy()
current_node = self.entry_point
print(f"\n{'='*70}")
print("GRAPH EXECUTION")
print(f"{'='*70}\n")
visited = []
while current_node and current_node != "END":
visited.append(current_node)
print(f"🔹 Node: {current_node}")
# Execute node
if current_node in self.nodes:
state = self.nodes[current_node](state)
# Determine next node
if current_node in self.conditional_edges:
# Use conditional logic
next_node = self.conditional_edges[current_node](state)
print(f" → Conditional: {next_node}")
elif current_node in self.edges:
# Use direct edge
next_node = self.edges[current_node]
print(f" → Next: {next_node}")
else:
# No more edges
next_node = "END"
current_node = next_node
print()
print(f"Path taken: {' → '.join(visited)}")
print(f"{'='*70}\n")
return state
print("SimpleGraph class defined")
Building a Research Agent with Graph¶
import random
# Define node functions
def plan_node(state: Dict) -> Dict:
"""Plan what to research"""
query = state['query']
thought = f"I need to research: {query}"
state['thoughts'].append(thought)
print(f" 💭 {thought}")
return state
def search_node(state: Dict) -> Dict:
"""Search for information"""
# Mock search
query = state['query']
results = [
f"Search result 1 for '{query}'",
f"Search result 2 for '{query}'",
f"Search result 3 for '{query}'"
]
observation = f"Found {len(results)} results"
state['observations'].append(observation)
print(f" 🔍 {observation}")
state['iteration'] += 1
return state
def summarize_node(state: Dict) -> Dict:
"""Summarize findings"""
summary = f"Summary of findings for: {state['query']}"
state['answer'] = summary
print(f" 📝 {summary}")
return state
def generate_node(state: Dict) -> Dict:
"""Generate answer without research"""
answer = f"Direct answer for: {state['query']}"
state['answer'] = answer
print(f" ✍️ {answer}")
return state
def review_node(state: Dict) -> Dict:
"""Review the answer"""
quality = random.choice(["good", "needs_improvement"])
state['quality'] = quality
print(f" ✓ Quality: {quality}")
return state
def revise_node(state: Dict) -> Dict:
"""Revise the answer"""
state['answer'] = f"Revised: {state['answer']}"
print(f" 🔄 Revised answer")
return state
# Build the graph
graph = SimpleGraph()
# Add nodes
graph.add_node("plan", plan_node)
graph.add_node("search", search_node)
graph.add_node("summarize", summarize_node)
graph.add_node("generate", generate_node)
graph.add_node("review", review_node)
graph.add_node("revise", revise_node)
# Add edges
graph.set_entry_point("plan")
# Conditional: Does query need research?
def should_research(state: Dict) -> str:
"""Decide if research is needed"""
# Simple heuristic: complex queries need research
needs_research = len(state['query'].split()) > 5
return "search" if needs_research else "generate"
graph.add_conditional_edge("plan", should_research)
# Search path
graph.add_edge("search", "summarize")
graph.add_edge("summarize", "review")
# Generate path
graph.add_edge("generate", "review")
# Review decision
def review_decision(state: Dict) -> str:
"""Decide if answer is good enough"""
return "END" if state.get('quality') == 'good' else "revise"
graph.add_conditional_edge("review", review_decision)
# Revise loops back to review
graph.add_edge("revise", "review")
print("Graph built successfully")
# Test with simple query (no research)
state1 = {
'query': 'Hello',
'thoughts': [],
'observations': [],
'answer': '',
'iteration': 0
}
result1 = graph.run(state1)
print(f"Final Answer: {result1['answer']}")
# Test with complex query (needs research)
state2 = {
'query': 'What are the latest developments in quantum computing and AI integration?',
'thoughts': [],
'observations': [],
'answer': '',
'iteration': 0
}
result2 = graph.run(state2)
print(f"Final Answer: {result2['answer']}")
print(f"Total Iterations: {result2['iteration']}")
Advanced: Parallel Execution¶
import concurrent.futures
class ParallelGraph(SimpleGraph):
"""Graph that can execute nodes in parallel"""
def add_parallel_nodes(self, name: str, nodes: List[Callable]):
"""Add nodes that execute in parallel"""
def parallel_executor(state: Dict) -> Dict:
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [executor.submit(node, state.copy()) for node in nodes]
results = [f.result() for f in concurrent.futures.as_completed(futures)]
# Merge results
for result in results:
for key, value in result.items():
if key in state and isinstance(state[key], list):
state[key].extend(value)
else:
state[key] = value
return state
self.add_node(name, parallel_executor)
# Example: Research multiple sources in parallel
def search_wikipedia(state: Dict) -> Dict:
state['observations'].append("Wikipedia results")
print(" 📚 Searched Wikipedia")
return state
def search_arxiv(state: Dict) -> Dict:
state['observations'].append("ArXiv results")
print(" 📄 Searched ArXiv")
return state
def search_news(state: Dict) -> Dict:
state['observations'].append("News results")
print(" 📰 Searched News")
return state
# Build parallel graph
parallel_graph = ParallelGraph()
parallel_graph.add_node("plan", plan_node)
parallel_graph.add_parallel_nodes("parallel_search", [
search_wikipedia,
search_arxiv,
search_news
])
parallel_graph.add_node("summarize", summarize_node)
parallel_graph.set_entry_point("plan")
parallel_graph.add_edge("plan", "parallel_search")
parallel_graph.add_edge("parallel_search", "summarize")
# Test parallel execution
state3 = {
'query': 'AI research trends',
'thoughts': [],
'observations': [],
'answer': ''
}
result3 = parallel_graph.run(state3)
print(f"\nObservations collected: {len(result3['observations'])}")
for obs in result3['observations']:
print(f" - {obs}")
Real LangGraph Example¶
Here’s how to use the actual LangGraph library:
# Example with real LangGraph (requires installation)
'''
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
# Define state
class AgentState(TypedDict):
messages: Annotated[List, operator.add]
query: str
answer: str
# Create graph
workflow = StateGraph(AgentState)
# Add nodes
workflow.add_node("plan", plan_node)
workflow.add_node("search", search_node)
workflow.add_node("generate", generate_node)
# Set entry point
workflow.set_entry_point("plan")
# Add edges
workflow.add_conditional_edges(
"plan",
should_research,
{
"search": "search",
"generate": "generate"
}
)
workflow.add_edge("search", "generate")
workflow.add_edge("generate", END)
# Compile
app = workflow.compile()
# Run
result = app.invoke({
"query": "What is LangGraph?",
"messages": [],
"answer": ""
})
'''
print("Real LangGraph example (commented - requires installation)")
Best Practices¶
1. State Design¶
Keep state minimal
Use typed dictionaries
Document state fields
Use operators for accumulation
2. Node Functions¶
One responsibility per node
Pure functions (no side effects)
Return modified state
Handle errors gracefully
3. Graph Structure¶
Start simple, add complexity gradually
Use conditional edges for branching
Avoid cycles (or limit iterations)
Document the workflow
4. Debugging¶
Log state at each node
Visualize the graph
Track execution path
Use breakpoints
Key Takeaways¶
✅ Graphs enable complex, stateful agent workflows
✅ Conditional edges allow branching logic
✅ Parallel execution improves performance
✅ State management is crucial for multi-step tasks
✅ LangGraph provides production-ready graph execution