LangGraph’s stateful agents are how you build AI systems that can actually do things, not just answer questions. They’re not just about chaining prompts; they’re about creating a loop where an AI can decide what to do next, do it, observe the result, and then decide again, potentially for many steps.
Let’s see this in action. Imagine an AI agent tasked with researching a topic, summarizing it, and then writing a blog post. Here’s a simplified LangGraph StateGraph that orchestrates this:
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class ResearchState(TypedDict):
topic: str
research_results: Annotated[list[str], operator.add]
summary: str
blog_post: str
def research_node(state: ResearchState):
# In a real scenario, this would call a tool to search the web.
# For demonstration, we'll return some fake results.
print("--> Researching...")
return {"research_results": [f"Result 1 for {state['topic']}", f"Result 2 for {state['topic']}"]}
def summarize_node(state: ResearchState):
print("--> Summarizing...")
# In a real scenario, this would use an LLM to summarize.
return {"summary": f"Summary of {state['topic']} based on results."}
def write_blog_post_node(state: ResearchState):
print("--> Writing blog post...")
# In a real scenario, this would use an LLM to write the post.
return {"blog_post": f"Blog post about {state['topic']}:\n{state['summary']}"}
builder = StateGraph(ResearchState)
builder.add_node("research", research_node)
builder.add_node("summarize", summarize_node)
builder.add_node("write_blog_post", write_blog_post_node)
builder.set_entry_point("research")
builder.add_edge("research", "summarize")
builder.add_edge("summarize", "write_blog_post")
builder.add_edge("write_blog_post", END)
graph = builder.compile()
# Run the graph
initial_state = {"topic": "LangGraph"}
result = graph.invoke(initial_state)
print(result)
This StateGraph defines a sequence of operations: research -> summarize -> write_blog_post. The ResearchState dictionary holds the data that flows between these steps. The Annotated[list[str], operator.add] for research_results is key: it tells LangGraph how to combine results if the research_node were called multiple times (it would append new results to the existing list).
The true power comes when you introduce conditional logic, allowing the agent to decide which node to go to next, or even to loop. For instance, if the initial research isn’t sufficient, the agent could decide to research more before summarizing. This is done by defining add_conditional_edges.
Let’s extend the example to include conditional logic. Suppose our agent needs to ensure it has enough research results before summarizing.
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class ResearchState(TypedDict):
topic: str
research_results: Annotated[list[str], operator.add]
summary: str
blog_post: str
needs_more_research: bool # New state variable
def research_node(state: ResearchState):
print("--> Researching...")
# Simulate getting more results if needed
if state.get("needs_more_research", False):
return {"research_results": [f"More result A for {state['topic']}", f"More result B for {state['topic']}"]}
else:
return {"research_results": [f"Initial result 1 for {state['topic']}", f"Initial result 2 for {state['topic']}"]}
def summarize_node(state: ResearchState):
print("--> Summarizing...")
return {"summary": f"Summary of {state['topic']} based on results."}
def write_blog_post_node(state: ResearchState):
print("--> Writing blog post...")
return {"blog_post": f"Blog post about {state['topic']}:\n{state['summary']}"}
def decide_research_needs(state: ResearchState) -> str:
print(f"--> Deciding research needs. Results count: {len(state.get('research_results', []))}")
if len(state.get('research_results', [])) < 3: # Arbitrary threshold
return "needs_more_research"
else:
return "summarize"
builder = StateGraph(ResearchState)
builder.add_node("research", research_node)
builder.add_node("summarize", summarize_node)
builder.add_node("write_blog_post", write_blog_post_node)
builder.set_entry_point("research")
# Use a conditional edge based on the decision function
builder.add_conditional_edges(
"research",
decide_research_needs,
{
"needs_more_research": "research", # Loop back to research if more is needed
"summarize": "summarize"
}
)
builder.add_edge("summarize", "write_blog_post")
builder.add_edge("write_blog_post", END)
graph = builder.compile()
# Run the graph
initial_state = {"topic": "LangGraph", "needs_more_research": True} # Start by needing more research
result = graph.invoke(initial_state)
print(result)
In this enhanced example, the decide_research_needs function checks the research_results. If there are fewer than 3 results, it returns "needs_more_research", causing the graph to loop back to the research node. Otherwise, it returns "summarize", moving the execution forward. This ability to self-correct and iterate is what makes LangGraph powerful for complex AI tasks. The needs_more_research flag is set in the initial state to demonstrate the loop.
The core idea is that each node in the graph represents a distinct function or tool call. The state dictionary is the shared memory, and the edges (regular or conditional) define the flow of control. The operator.add for research_results is a simple example of an Annotated type that specifies how to merge values if a node is executed multiple times. LangGraph supports more complex merge strategies for various data types.
The most surprising thing about LangGraph is how it enables you to implement sophisticated agentic behavior using a surprisingly simple, declarative graph structure. You’re not writing imperative loops and complex state management code; you’re defining the nodes and the transitions between them, and LangGraph handles the execution and state persistence. This abstraction allows you to focus on the AI’s capabilities and decision-making logic, rather than the boilerplate of orchestrating calls.
The Annotated type hint is crucial for defining how state variables are updated when a node is executed. By default, if a key is present in the returned dictionary from a node, it overwrites the existing value in the state. However, using Annotated with operators like operator.add (for lists or numbers) or operator.xor (for sets) allows for custom merge strategies. This means you can build agents that accumulate information, maintain unique sets of observations, or perform other complex state manipulations without explicit manual merging logic in your nodes.
The next concept you’ll likely encounter is implementing advanced persistence and checkpointing for these stateful agents.