Complete LangGraph Tutorial: Building Complex AI Agents

LangGraph is a library from LangChain for building stateful, multi-actor applications with Large Language Models (LLMs). With LangGraph, you can create complex AI agents with flexible control flow, state management, and the ability to run multiple agents simultaneously.

What is LangGraph?

LangGraph extends LangChain by providing:

• Stateful graphs: Store and manage state between steps
• Cyclic flows: Support for loops and conditionals
• Human-in-the-loop: Manual interaction during workflow
• Persistence: Save state for resume later
• Streaming: Real-time streaming responses

• Building AI agents with multi-step reasoning
• Workflows requiring conditionals and loops
• Applications needing human approval
• Multi-agent systems
• Complex RAG pipelines

Installation

pip install langgraph langchain langchain-openai

Basic Concepts

1. Graph Components

LangGraph consists of:

• State: Data managed throughout execution
• Nodes: Functions that process and modify state
• Edges: Connections between nodes (conditional or fixed)

2. Basic Graph Structure

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

Define state
class State(TypedDict):
    messages: list
    currentstep: str


Define nodes
def nodea(state: State) -> State:
    state["messages"].append("Processed by Node A")
    state["currentstep"] = "a"

    return state

def nodeb(state: State) -> State:
    state["messages"].append("Processed by Node B")
    state["currentstep"] = "b"

    return state

Build graph
graph = StateGraph(State)
graph.addnode("nodea", nodea)
graph.addnode("nodeb", nodeb)


Add edges
graph.addedge(START, "nodea")

graph.addedge("nodea", "nodeb")
graph.addedge("nodeb", END)

Compile
app = graph.compile()

Run
result = app.invoke({"messages": [], "currentstep": ""})

print(result)

from typing import Annotated
from typingextensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages

from langchainopenai import ChatOpenAI
from langchaincore.messages import HumanMessage, AIMessage


State with message history
class ChatState(TypedDict):
    messages: Annotated[list, addmessages]

Initialize LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

Chatbot node
def chatbot(state: ChatState) -> ChatState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

Build graph
graph = StateGraph(ChatState)
graph.addnode("chatbot", chatbot)

graph.addedge(START, "chatbot")
graph.addedge("chatbot", END)


app = graph.compile()

Interactive chat
def chat(userinput: str, history: list = None):
    if history is None:
        history = []

    history.append(HumanMessage(content=userinput))

    result = app.invoke({"messages": history})

    return result["messages"]

Usage
messages = chat("Hello, who are you?")
print(messages[-1].content)

messages = chat("What can you do?", messages)
print(messages[-1].content)

from typing import Annotated
from typingextensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages

from langgraph.prebuilt import ToolNode, toolscondition
from langchainopenai import ChatOpenAI

from langchaincore.tools import tool

Define tools
@tool
def searchweb(query: str) -> str:

    """Search the web for information."""
    # Simulate web search
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Calculate a mathematical expression."""
    try:
        result = eval(expression)
        return f"Result: {result}"
    except:
        return "Error calculating expression"

@tool
def getweather(city: str) -> str:
    """Get current weather for a city."""
    # Simulate weather API
    return f"Weather in {city}: 25°C, Sunny"

tools = [searchweb, calculator, getweather]

State
class AgentState(TypedDict):
    messages: Annotated[list, addmessages]


LLM with tools
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llmwithtools = llm.bindtools(tools)

Agent node
def agent(state: AgentState) -> AgentState:
    response = llmwithtools.invoke(state["messages"])
    return {"messages": [response]}

Build graph
graph = StateGraph(AgentState)

Add nodes
graph.addnode("agent", agent)

graph.addnode("tools", ToolNode(tools))

Add edges
graph.addedge(START, "agent")

graph.addconditionaledges(

    "agent",
    toolscondition,  # Routes to "tools" if tool call, else END
)
graph.addedge("tools", "agent")


app = graph.compile()

Test
from langchaincore.messages import HumanMessage

result = app.invoke({
    "messages": [HumanMessage(content="What's the weather in Jakarta?")]
})
print(result["messages"][-1].content)

result = app.invoke({
    "messages": [HumanMessage(content="Calculate 25  4 + 100")]

})
print(result["messages"][-1].content)

from typing import Annotated, Literal
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages
from langchainopenai import ChatOpenAI

from langchaincore.tools import tool
from langchaincore.messages import HumanMessage, AIMessage, ToolMessage


Tools
@tool
def searchdatabase(query: str) -> str:
    """Search internal database for information."""
    # Simulate database search
    data = {
        "producta": "Price: $100, Stock: 50",

        "productb": "Price: $200, Stock: 25",
        "usercount": "Total users: 10,000"

    }
    for key, value in data.items():
        if key in query.lower():
            return value
    return "No results found"

@tool
def sendemail(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    return f"Email sent to {to} with subject: {subject}"

tools = [searchdatabase, sendemail]

State
class ReActState(TypedDict):
    messages: Annotated[list, addmessages]

    iteration: int

LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llmwithtools = llm.bindtools(tools)

Nodes
def reasoning(state: ReActState) -> ReActState:
    """Agent reasoning step"""
    response = llmwithtools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1
    }

def acting(state: ReActState) -> ReActState:
    """Execute tool calls"""
    lastmessage = state["messages"][-1]

    toolresults = []

    for toolcall in lastmessage.toolcalls:

        toolname = toolcall["name"]

        toolargs = toolcall["args"]


        # Find and execute tool
        for t in tools:
            if t.name == toolname:
                result = t.invoke(toolargs)

                toolresults.append(
                    ToolMessage(content=result, toolcallid=toolcall["id"])

                )
                break

    return {"messages": toolresults}

def shouldcontinue(state: ReActState) -> Literal["acting", "end"]:

    """Determine if we should continue or end"""
    lastmessage = state["messages"][-1]

    # Check iteration limit
    if state.get("iteration", 0) >= 5:
        return "end"

    # Check if there are tool calls
    if hasattr(lastmessage, "toolcalls") and lastmessage.toolcalls:
        return "acting"

    return "end"

Build graph
graph = StateGraph(ReActState)

graph.addnode("reasoning", reasoning)

graph.addnode("acting", acting)

graph.addedge(START, "reasoning")

graph.addconditionaledges(

    "reasoning",
    shouldcontinue,
    {"acting": "acting", "end": END}
)
graph.addedge("acting", "reasoning")


app = graph.compile()

Test
result = app.invoke({
    "messages": [HumanMessage(content="Search for producta info and then send it to user@example.com")],
    "iteration": 0
})

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

from typing import Annotated, Literal
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages
from langchainopenai import ChatOpenAI

from langchaincore.messages import HumanMessage, AIMessage

State
class MultiAgentState(TypedDict):
    messages: Annotated[list, addmessages]

    nextagent: str

Agents
def researcher(state: MultiAgentState) -> MultiAgentState:
    """Research agent - gathers information"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    system = """You are a research agent. Your job is to gather and analyze information.
    Be thorough but concise in your findings."""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=f"[Researcher]: {response.content}")]}

def writer(state: MultiAgentState) -> MultiAgentState:
    """Writer agent - creates content"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

    system = """You are a writer agent. Your job is to create well-written content
    based on the research provided. Be creative and engaging."""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=f"[Writer]: {response.content}")]}

def reviewer(state: MultiAgentState) -> MultiAgentState:
    """Reviewer agent - reviews and improves content"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    system = """You are a reviewer agent. Your job is to review content and provide
    constructive feedback or improvements."""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=f"[Reviewer]: {response.content}")]}

def supervisor(state: MultiAgentState) -> MultiAgentState:
    """Supervisor - decides which agent to call next"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    system = """You are a supervisor managing a team of agents:
researcher: gathers information
writer: creates content
reviewer: reviews content

    Based on the conversation, decide which agent should work next.
    Respond with ONLY one of: researcher, writer, reviewer, or FINISH"""

    messages = [{"role": "system", "content": system}] + [
        {"role": m.type, "content": m.content} for m in state["messages"]
    ]

    response = llm.invoke(messages)
    nextagent = response.content.strip().lower()


    return {"nextagent": nextagent}


def routesupervisor(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]:
    """Route based on supervisor decision"""
    nextagent = state.get("nextagent", "").lower()

    if "finish" in nextagent:

        return "end"
    elif "researcher" in nextagent:
        return "researcher"
    elif "writer" in nextagent:

        return "writer"
    elif "reviewer" in nextagent:
        return "reviewer"
    else:
        return "end"

Build graph
graph = StateGraph(MultiAgentState)

graph.addnode("supervisor", supervisor)

graph.addnode("researcher", researcher)
graph.addnode("writer", writer)

graph.addnode("reviewer", reviewer)

graph.addedge(START, "supervisor")

graph.addconditionaledges(

    "supervisor",
    routesupervisor,
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "end": END
    }
)
graph.addedge("researcher", "supervisor")

graph.addedge("writer", "supervisor")
graph.addedge("reviewer", "supervisor")


app = graph.compile()

Test
result = app.invoke({
    "messages": [HumanMessage(content="Write a blog post about AI in healthcare")],
    "nextagent": ""
})

for msg in result["messages"]:
    print(f"\n{msg.content}\n{'='50}")

Human-in-the-Loop

1. Interrupt for Approval

from typing import Annotated
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import addmessages
from langgraph.checkpoint.memory import MemorySaver
from langchainopenai import ChatOpenAI

from langchaincore.messages import HumanMessage

State
class ApprovalState(TypedDict):
    messages: Annotated[list, addmessages]

    draft: str
    approved: bool

Nodes
def createdraft(state: ApprovalState) -> ApprovalState:
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(state["messages"])
    return {"draft": response.content, "approved": False}

def processapproval(state: ApprovalState) -> ApprovalState:

    # This node processes after human approval
    return {"messages": [AIMessage(content=f"Final approved content: {state['draft']}")]}

def checkapproval(state: ApprovalState) -> str:
    if state.get("approved", False):
        return "process"
    return "wait"

Build graph
graph = StateGraph(ApprovalState)

graph.addnode("createdraft", createdraft)

graph.addnode("processapproval", processapproval)

graph.addedge(START, "createdraft")
graph.addconditionaledges(
    "createdraft",

    checkapproval,
    {"process": "processapproval", "wait": END}

)
graph.addedge("processapproval", END)


Compile with checkpointer for persistence
memory = MemorySaver()
app = graph.compile(checkpointer=memory, interruptbefore=["processapproval"])


Usage with human approval
config = {"configurable": {"threadid": "1"}}

Step 1: Create draft
result = app.invoke(
    {"messages": [HumanMessage(content="Write an email to client about project delay")]},
    config
)
print("Draft created:", result["draft"])

Step 2: Human reviews and approves
In real app, this would be UI interaction
print("\n[Human approves the draft]\n")

Step 3: Continue with approval
result = app.invoke(
    {"approved": True},
    config
)
print("Final:", result["messages"][-1].content)

Persistence and Checkpointing

1. Memory Saver (In-Memory)

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

Run with threadid

config = {"configurable": {"threadid": "user123"}}

result = app.invoke({"messages": [HumanMessage(content="Hello")]}, config)

Continue conversation
result = app.invoke({"messages": [HumanMessage(content="What did I say?")]}, config)

from langgraph.checkpoint.sqlite import SqliteSaver

Create SQLite checkpointer
with SqliteSaver.fromconnstring(":memory:") as checkpointer:

    app = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"threadid": "session1"}}

    result = app.invoke({"messages": [HumanMessage(content="Hello")]}, config)

from langgraph.checkpoint.postgres import PostgresSaver

connectionstring = "postgresql://user:password@localhost/langgraph"

with PostgresSaver.fromconnstring(connectionstring) as checkpointer:

    app = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"threadid": "persistentsession"}}

    result = app.invoke({"messages": [HumanMessage(content="Hello")]}, config)

from langchaincore.messages import HumanMessage

Stream all events
for event in app.stream({"messages": [HumanMessage(content="Tell me a story")]}):
    print(event)
    print("---")

2. Stream with Mode

# Stream values (state after each node)
for state in app.stream(
    {"messages": [HumanMessage(content="Hello")]},
    streammode="values"

):
    print(state["messages"][-1].content)

Stream updates (only changes)
for update in app.stream(
    {"messages": [HumanMessage(content="Hello")]},
    streammode="updates"
):
    print(update)

3. Async Streaming

import asyncio

async def streamresponse():

    async for event in app.astream(
        {"messages": [HumanMessage(content="Hello")]}
    ):
        print(event)

asyncio.run(streamresponse())

RAG with LangGraph

from typing import Annotated, List
from typingextensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langchainopenai import ChatOpenAI, OpenAIEmbeddings
from langchaincommunity.vectorstores import FAISS

from langchaincore.messages import HumanMessage

State
class RAGState(TypedDict):
    question: str
    context: List[str]
    answer: str

Setup vector store
embeddings = OpenAIEmbeddings()
documents = [
    "LangGraph is a library for building stateful AI applications.",
    "It supports cyclic flows and human-in-the-loop interactions.",
    "LangGraph extends LangChain with graph-based workflows.",
]
vectorstore = FAISS.fromtexts(documents, embeddings)

retriever = vectorstore.asretriever(k=2)

Nodes
def retrieve(state: RAGState) -> RAGState:
    """Retrieve relevant documents"""
    docs = retriever.invoke(state["question"])
    return {"context": [doc.pagecontent for doc in docs]}


def generate(state: RAGState) -> RAGState:
    """Generate answer based on context"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    context = "\n".join(state["context"])
    prompt = f"""Answer the question based on the context.

Context:
{context}

Question: {state["question"]}

Answer:"""

    response = llm.invoke(prompt)
    return {"answer": response.content}

Build graph
graph = StateGraph(RAGState)

graph.addnode("retrieve", retrieve)
graph.addnode("generate", generate)


graph.addedge(START, "retrieve")
graph.addedge("retrieve", "generate")

graph.addedge("generate", END)

app = graph.compile()

Test
result = app.invoke({"question": "What is LangGraph?"})
print(f"Answer: {result['answer']}")

Best Practices

1. Error Handling

from langgraph.graph import StateGraph

def safenode(state):

    try:
        # Your logic here
        result = riskyoperation()
        return {"result": result, "error": None}
    except Exception as e:
        return {"result": None, "error": str(e)}

def errorrouter(state):

    if state.get("error"):
        return "handleerror"
    return "continue"

2. Timeout and Retry

import asyncio
from tenacity import retry, stopafterattempt, waitexponential


@retry(stop=stopafterattempt(3), wait=waitexponential(multiplier=1, min=4, max=10))
def nodewithretry(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

3. Logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

def logged_node(state):
    logger.info(f"Processing state: {state}")
    result = process(state)
    logger.info(f"Result: {result}")
    return result

Conclusion

LangGraph provides a powerful framework for building complex AI agents with:

Key takeaways:

• Use TypedDict for type-safe state
• Leverage conditional edges for dynamic routing
• Implement checkpointing for production
• Use streaming for real-time responses
• Handle errors gracefully with fallbacks