Build an AI Agent with LangChain: Complete LangGraph Tutorial

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.

Frequently Asked Questions

Why use LangGraph instead of the older LangChain AgentExecutor?

LangGraph provides explicit state management, full execution visibility, human-in-the-loop checkpoints, conditional branching, and persistence across sessions. AgentExecutor is a black box. LangGraph is now LangChain's recommended production agent approach.

What is agent state in LangGraph?

A TypedDict that holds all agent information — messages, intermediate results, task context. Passed to and from every node. Explicit, typed, serializable. Enables checkpointing, testing individual nodes, and debugging by inspecting state at each step.

How do I add human-in-the-loop?

Compile with interrupt_before=['node_name'] and a checkpointer. Agent pauses before that node, returns current state for review. Call app.invoke(None, thread) to resume. Use app.update_state() to inject human feedback before resuming.

How do I persist agent state across sessions?

Use SqliteSaver (single-process) or PostgresSaver (multi-process) as checkpointer. Each session has a thread_id. Resuming: pass the same thread_id to load the last checkpoint and continue.

How do I handle errors in a LangGraph agent?

Use try/except in node functions, return error info in state, route to error-handling nodes via conditional edges. Set recursion_limit to prevent infinite loops. Implement max_retries for tool calls. Log every state transition for debugging.