LangGraph Tutorial: Stateful AI Agents

Why LangGraph?

Simple LLM chains work for linear tasks. Real agents need:

Persistent state across multiple steps

Conditional branching (different actions based on results)

Loops (retry until success)

Human-in-the-loop interruptions

Parallel execution of independent tasks

LangGraph models agents as directed graphs where nodes are functions and edges define flow.

Installation

bash
pip install langgraph langchain-openai tavily-python

Core Concepts

State: A TypedDict that persists throughout the graph execution. Nodes: Python functions that read/modify state. Edges: Connections defining flow. Can be conditional. Checkpointer: Persists state between runs (enables human-in-the-loop).

Building a Research Agent

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
1. Define state
class ResearchState(TypedDict):
    query: str
    search_results: List[dict]
    analysis: str
    final_report: str
    iterations: int
2. Define tools
tools = [TavilySearchResults(max_results=5)]
llm = ChatOpenAI(model='gpt-4o').bind_tools(tools)
tool_node = ToolNode(tools)
3. Define nodes
def search_web(state: ResearchState) -> ResearchState:
    response = llm.invoke(f'Search for information about: {state["query"]}')
    return {'iterations': state.get('iterations', 0) + 1}
def analyze_results(state: ResearchState) -> ResearchState:
    results_text = str(state.get('search_results', []))
    analysis = llm.invoke(f'Analyze these search results: {results_text}')
    return {'analysis': analysis.content}
def write_report(state: ResearchState) -> ResearchState:
    report = llm.invoke(
        f'Write a comprehensive report about {state["query"]}'
        f'based on: {state["analysis"]}'
    )
    return {'final_report': report.content}
4. Build graph
workflow = StateGraph(ResearchState)
workflow.add_node('search', search_web)
workflow.add_node('analyze', analyze_results)
workflow.add_node('write', write_report)
workflow.set_entry_point('search')
workflow.add_edge('search', 'analyze')
workflow.add_edge('analyze', 'write')
workflow.add_edge('write', END)
agent = workflow.compile()

Adding Conditional Logic

python
def should_search_more(state: ResearchState) -> str:
    if state.get('iterations', 0) < 3 and len(state.get('analysis', '')) < 500:
        return 'search_more'
    return 'write_report'
workflow.add_conditional_edges(
    'analyze',
    should_search_more,
    {
        'search_more': 'search',
        'write_report': 'write'
    }
)

Human-in-the-Loop

python
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(':memory:')
agent = workflow.compile(checkpointer=memory, interrupt_before=['write'])
Run until interrupt
config = {'configurable': {'thread_id': 'research-1'}}
result = agent.invoke({'query': 'AI regulation 2026'}, config=config)
Human reviews analysis, then continues
print('Analysis for review:', result['analysis'])
agent.invoke(None, config=config)  # Resume from checkpoint

Parallel Execution

python
from langgraph.graph import START
Run multiple searches in parallel
workflow.add_edge(START, 'search_news')
workflow.add_edge(START, 'search_research')
workflow.add_edge(START, 'search_examples')
All three run simultaneously, then merge
workflow.add_edge(['search_news', 'search_research', 'search_examples'], 'merge')

When to Use LangGraph

Multi-step workflows with conditional branching

Long-running tasks requiring state persistence

Human-in-the-loop approval workflows

Multi-agent coordination

Tasks requiring retry logic

Not needed for: simple single-shot queries, basic RAG retrieval.