Why use LangGraph instead of the older LangChain AgentExecutor?

The older LangChain AgentExecutor is a black box — you can't control execution flow, add custom logic between steps, or implement human-in-the-loop. LangGraph treats the agent as an explicit state machine with nodes (steps) and edges (transitions). Benefits: full visibility into agent state at every step, the ability to add human approval checkpoints, conditional branching based on results, custom error recovery, parallel execution, and persistence across sessions. LangGraph is now LangChain's recommended approach for all agent development. AgentExecutor still works but lacks these production-critical controls.

What is agent state in LangGraph?

Agent state is a TypedDict (or Pydantic model) that holds all information the agent needs across steps. It's passed to every node, modified, and returned. Unlike implicit context in older agents, state is explicit, typed, and serializable. Typical fields: messages (conversation history), intermediate_steps (tool calls and results), task_context (current goal), output (final result). State enables: resuming from checkpoints after failure, human review at any point, testing individual nodes, debugging by inspecting state at each step. Reducers (like operator.add for lists) control how state fields are updated when multiple nodes write to the same key.

How do I add human-in-the-loop to a LangGraph agent?

Use interrupt_before or interrupt_after when compiling the graph, plus a checkpointer for state persistence. compile(checkpointer=checkpointer, interrupt_before=['expensive_node']) pauses execution before that node. The graph returns the current state for human review. After review, call app.invoke(None, thread_config) to resume from the checkpoint. For more control: use Command(resume=value) to pass human input back into the graph. This is critical for production agents where you need approval before: sending emails, making API calls with side effects, executing code in production, or taking any irreversible action.

How do I persist agent state across sessions?

Use a LangGraph checkpointer: MemorySaver for in-process development, SqliteSaver for persistent single-process storage, PostgresSaver for production multi-process deployments. The checkpointer saves state at every step using a thread_id. Resuming a previous session: app.invoke(None, {configurable: {thread_id: 'previous-id'}}) loads the last checkpoint and continues. This enables: long-running tasks that span hours or days, recovering from crashes without restarting, multi-user applications where each user has their own thread, and human-in-the-loop where the human reviews asynchronously.

How do I handle errors in a LangGraph agent?

LangGraph handles errors through: try/except in node functions (catch and return error state), conditional edges that route to error-handling nodes when output signals failure, retry logic using recursion limits (set_recursion_limit), and fallback nodes. Best practice: return structured error information in state rather than raising exceptions, use conditional edges to route to recovery nodes, implement max_retries for tool calls, and add a 'stuck' detection (if the same action is repeated N times, escalate to human). For production: set recursion_limit to prevent infinite loops, implement timeouts on tool calls, and log every state transition for debugging.

Build an AI Agent with LangChain: Complete LangGraph Tutorial

The first agent I built with LangChain's old AgentExecutor worked for simple tasks but was a black box I couldn't debug or extend. When it went wrong, I had no way to see where or redirect it.

LangGraph changed this. By treating agents as explicit state machines, every step is visible, testable, and controllable. You can pause at any point, inspect state, add human review, and resume. This turns agents from research projects into production systems.

Setup

pip install langchain langchain-openai langgraph

# For persistence
pip install langgraph-checkpoint-sqlite  # Development
# pip install langgraph-checkpoint-postgres  # Production

Part 1: Simple ReAct Agent with LangGraph

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

# Define tools
@tool
def search(query: str) -> str:
    """Search for current information on the web."""
    # In production: use SerpAPI, Tavily, or DuckDuckGo
    return f"Search results for '{query}': [simulated results]"

@tool
def calculate(expression: str) -> str:
    """Calculate a mathematical expression."""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather in {city}: 22°C, partly cloudy"

tools = [search, calculate, get_weather]
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Create agent with memory
memory = MemorySaver()
agent = create_react_agent(model, tools, checkpointer=memory)

# Run agent
config = {"configurable": {"thread_id": "user-session-1"}}

result = agent.invoke(
    {"messages": [("human", "What's the weather in Tokyo and what is 15% of 2500?")]},
    config
)

for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

# Continue same conversation (memory persists)
result2 = agent.invoke(
    {"messages": [("human", "How does that compare to Paris?")]},
    config  # Same thread_id = same conversation
)

Part 2: Custom Agent with LangGraph State Machine

from typing import TypedDict, Annotated
import operator
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import json

# Define typed state
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    task_complete: bool
    iterations: int
    final_answer: str

# LLM with tools bound
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [search, calculate, get_weather]
llm_with_tools = llm.bind_tools(tools)

# Node: LLM reasoning step
def agent_node(state: AgentState) -> AgentState:
    messages = state["messages"]
    
    # Add system message if first iteration
    if state["iterations"] == 0:
        messages = [
            SystemMessage(content="You are a helpful assistant. Use tools when needed. "
                                  "When you have a complete answer, provide it clearly.")
        ] + messages
    
    response = llm_with_tools.invoke(messages)
    
    return {
        "messages": [response],
        "iterations": state["iterations"] + 1,
        "task_complete": False,
        "final_answer": ""
    }

# Node: Execute tools
tool_node = ToolNode(tools)

# Edge: Decide what to do next
def should_continue(state: AgentState) -> str:
    messages = state["messages"]
    last_message = messages[-1]
    
    # Check for max iterations (prevent infinite loops)
    if state["iterations"] >= 10:
        return "end"
    
    # If LLM called tools, execute them
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # LLM produced final answer
    return "end"

# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)

workflow.add_edge("tools", "agent")  # After tools, return to agent

app = workflow.compile()

# Run
initial_state = {
    "messages": [HumanMessage(content="What is the population of Tokyo divided by 1000?")],
    "task_complete": False,
    "iterations": 0,
    "final_answer": ""
}

result = app.invoke(initial_state)

# Print execution trace
for msg in result["messages"]:
    print(f"\n[{msg.type}] {msg.content[:200]}")

Part 3: Multi-Step Research Agent

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class ResearchState(TypedDict):
    topic: str
    research_queries: List[str]
    search_results: Annotated[List[str], operator.add]
    analysis: str
    report: str
    quality_check: str
    approved: bool

llm = ChatOpenAI(model="gpt-4o-mini")

def plan_research(state: ResearchState) -> ResearchState:
    """Generate research questions."""
    response = llm.invoke([
        SystemMessage(content="Generate 3 specific search queries for researching this topic."),
        HumanMessage(content=f"Topic: {state['topic']}")
    ])
    
    queries = [q.strip() for q in response.content.split("\n") if q.strip()][:3]
    return {"research_queries": queries}

def execute_research(state: ResearchState) -> ResearchState:
    """Execute each search query."""
    results = []
    for query in state["research_queries"]:
        result = search.invoke(query)
        results.append(f"Query: {query}\nResult: {result}")
    
    return {"search_results": results}

def analyze_results(state: ResearchState) -> ResearchState:
    """Analyze gathered research."""
    context = "\n\n".join(state["search_results"])
    
    response = llm.invoke([
        SystemMessage(content="Analyze these research results and identify key findings."),
        HumanMessage(content=f"Topic: {state['topic']}\n\nResearch:\n{context}")
    ])
    
    return {"analysis": response.content}

def write_report(state: ResearchState) -> ResearchState:
    """Write the final report."""
    response = llm.invoke([
        SystemMessage(content="Write a concise, well-structured report based on this analysis."),
        HumanMessage(content=f"Topic: {state['topic']}\n\nAnalysis:\n{state['analysis']}")
    ])
    
    return {"report": response.content}

def quality_check(state: ResearchState) -> ResearchState:
    """Evaluate report quality."""
    response = llm.invoke([
        SystemMessage(content="Rate this report 1-10 and identify any gaps. Be specific."),
        HumanMessage(content=f"Topic: {state['topic']}\n\nReport:\n{state['report']}")
    ])
    
    quality = response.content
    score_str = [c for c in quality if c.isdigit()]
    score = int(score_str[0]) if score_str else 5
    
    return {"quality_check": quality, "approved": score >= 7}

def should_revise(state: ResearchState) -> str:
    """Decide: done or needs revision."""
    return "done" if state["approved"] else "revise"

# Build research workflow
research_graph = StateGraph(ResearchState)
research_graph.add_node("plan", plan_research)
research_graph.add_node("research", execute_research)
research_graph.add_node("analyze", analyze_results)
research_graph.add_node("write", write_report)
research_graph.add_node("check", quality_check)

research_graph.set_entry_point("plan")
research_graph.add_edge("plan", "research")
research_graph.add_edge("research", "analyze")
research_graph.add_edge("analyze", "write")
research_graph.add_edge("write", "check")
research_graph.add_conditional_edges(
    "check",
    should_revise,
    {"done": END, "revise": "analyze"}  # Loop back to analyze if quality is low
)

research_app = research_graph.compile()

result = research_app.invoke({
    "topic": "Python asyncio best practices",
    "research_queries": [],
    "search_results": [],
    "analysis": "",
    "report": "",
    "quality_check": "",
    "approved": False
})

print(f"Final Report:\n{result['report']}")
print(f"\nQuality Check:\n{result['quality_check']}")

Part 4: Human-in-the-Loop

from langgraph.checkpoint.sqlite import SqliteSaver

# Compile with checkpointer for persistent state
with SqliteSaver.from_conn_string("agent_checkpoints.db") as checkpointer:
    agent_with_hitl = research_graph.compile(
        checkpointer=checkpointer,
        interrupt_before=["write"]  # Pause before writing the report
    )
    
    thread = {"configurable": {"thread_id": "research-001"}}
    
    # Start execution — will pause before "write" node
    result = agent_with_hitl.invoke(
        {"topic": "LLM evaluation methods", "research_queries": [], "search_results": [],
         "analysis": "", "report": "", "quality_check": "", "approved": False},
        thread
    )
    
    print("Current analysis (before writing report):")
    print(result["analysis"])
    
    # Human reviews the analysis and decides
    user_input = input("\nApprove analysis? (y/n/modify): ").strip().lower()
    
    if user_input == "y":
        # Resume execution
        final_result = agent_with_hitl.invoke(None, thread)
        print("\nFinal Report:")
        print(final_result["report"])
    
    elif user_input == "modify":
        # Inject human feedback into state
        feedback = input("Your modification: ")
        # Update state with human input, then resume
        agent_with_hitl.update_state(
            thread,
            {"analysis": result["analysis"] + f"\n\nHuman note: {feedback}"}
        )
        final_result = agent_with_hitl.invoke(None, thread)
        print("\nFinal Report (with modifications):")
        print(final_result["report"])

Part 5: Streaming Agent Execution

# Stream events for real-time monitoring
for event in app.stream(
    {"messages": [HumanMessage(content="Research the best Python ORMs")]},
    stream_mode="updates"
):
    for node, updates in event.items():
        print(f"\n[Node: {node}]")
        if "messages" in updates:
            for msg in updates["messages"]:
                print(f"  {msg.type}: {str(msg.content)[:200]}")

Conclusion

LangGraph's explicit state machine approach makes agents debuggable, testable, and production-ready. The graph structure turns an opaque agent loop into a visible workflow where you can inspect state at every step, add human review at key points, and implement proper error recovery.

For combining multiple LangGraph agents into multi-agent systems, see our CrewAI tutorial. For the OpenAI Assistants API as an alternative managed agent platform, see our OpenAI Assistants guide.