Codecademy Logo

Introduction to LangGraph

Related learning

  • Skill path

    Learn LangChain

    Build powerful AI applications using LangChain and LangGraph.
    • Includes 3 Courses
    • With Certificate
    • Intermediate.
      4 hours

LangGraph Framework

LangGraph is a framework built on top of LangChain for creating stateful, multi-step workflows as directed graphs, where each step is a node and the connections between steps are edges.

StateGraph Class

The StateGraph class is used to create graph workflows in LangGraph. You initialize it with a state schema (a Python TypedDict) that specifies the names and types of all shared fields. Each node in the graph reads from and writes to this shared state.

from langgraph.graph import StateGraph
from typing_extensions import TypedDict


# Define the state schema with all shared fields
class SupportTicketState(TypedDict):
    issue: str
    analysis: str
    resolution: str


# Initialize the graph with the state schema
builder = StateGraph(SupportTicketState)

START and END Constants

In LangGraph, START and END are special constants representing the entry and exit points of a graph. START links to the first node to process, whereas END signifies that execution is finished and no further nodes will run.

from langgraph.graph import START, END


builder.add_edge(START, "analyze")    # Entry point
builder.add_edge("respond", END)      # Exit point

Nodes with add_node()

A node is a Python function registered with add_node("name", function). Each node receives the current state as input and returns a dictionary containing only the fields it wants to update. LangGraph then automatically merges these updates into the shared state.

def analyze_issue(state: SupportTicketState):
    """Analyze the support issue and return findings."""
    result = llm.invoke(
        f"Analyze this support issue: {state['issue']}"
    )
    # Return only the field being updated
    return {"analysis": result.content}


builder.add_node("analyze", analyze_issue)

Edges with add_edge()

In LangGraph, the add_edge(source, target) method establishes a fixed connection between two nodes, called an edge, so that when the source node finishes, the target node executes next. 

# Define the execution order
builder.add_edge(START, "analyze")    
builder.add_edge("analyze", "respond")
builder.add_edge("respond", END)

Compiling the Graph

The compile() method in LangGraph validates the graph structure and returns an executable object that can be run with an initial state dictionary using the .invoke() method.

# Compile the graph into a runnable
support_pipeline = builder.compile()


# Run the graph with an initial state
result = support_pipeline.invoke({
    "issue": "Unable to reset password after multiple attempts",
    "analysis": "",
    "resolution": ""
})

Reducers in LangGraph

A reducer determines how a state field updates when a node provides a value: by default, values overwrite the field, but using Annotated[list, operator.add] changes this behavior to append new items instead of replacing them.

from typing import Annotated
import operator


class OrderState(TypedDict):
    order_id: str
    status: str
    # This field accumulates entries from all nodes
    audit_trail: Annotated[list, operator.add]

Conditional Routing

The add_conditional_edges() method links a source node to a routing function that evaluates the current state and returns the name of the next node to execute. This allows the graph to follow different paths during runtime based on the data.

from typing import Literal


# Define the routing function
def route_by_priority(state) -> Literal["escalate", "standard_response"]:
    """Choose the next node based on ticket priority."""
    if state["priority"] == "urgent":
        return "escalate"
    return "standard_response"


# Add conditional edges
builder.add_conditional_edges("analyze", route_by_priority)

Cycles for Iterative Execution

A cycle in LangGraph happens when an edge links a downstream node back to an upstream node, forming a loop. This allows for iterative execution, where results are improved repeatedly until a condition is satisfied, such as a quality threshold or maximum number of iterations.

# Conditional edge decides whether to continue looping
builder.add_conditional_edges(
    "review", should_refine,
    {"refine": "refine", "done": END}
)


# This edge creates the cycle back to review
builder.add_edge("refine", "review")

recursion_limit in LangGraph

In LangGraph, recursion_limit is a configuration parameter passed to .invoke() that sets the maximum number of node executions, preventing infinite loops by raising an error when the limit is exceeded.

# Set a safety limit of 10 node executions
result = support_pipeline.invoke(
    initial_state,
    config={"recursion_limit": 10}
)

LangGraph MemorySaver

In LangGraph, MemorySaver is an in-memory checkpointer that preserves the graph’s state during execution. Passing a checkpointer to compile() enables the graph to be paused and resumed later.

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Human in the Loop

In LangGraph, interrupt() pauses execution within a node and provides information for human review. When the human responds, Command(resume=...) sends their input back to the paused node, while Command(goto=...) directs execution to a specific next node based on that decision.

from langgraph.types import interrupt, Command


def manager_approval(state) -> Command[Literal["process", "reject"]]:
    """Pause for manager review before proceeding."""
    # Pause and show the request to a human
    decision = interrupt({
        "request": state["request"],
        "instruction": "Reply: 'approve' or 'reject: <reason>'"
    })
    
    # Route based on the human's decision
    if decision.startswith("approve"):
        return Command(goto="process")
    reason = decision.replace("reject:", "").strip()
    return Command(goto="reject", update={"reason": reason})

MessagesState in LangGraph

In LangGraph, MessagesState is a built-in state schema that includes a messages list, managed by the add_messages reducer. It automatically collects the entire conversation history across turns, storing message objects such as HumanMessage and AIMessage.

from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import HumanMessage


def chatbot_node(state: MessagesState):
    """Generate a response based on conversation history."""
    response = llm_with_tools.invoke(state["messages"])
    # Returning a list appends to the conversation
    return {"messages": [response]}


builder = StateGraph(MessagesState)

Tool Binding

In LangGraph, ToolNode is a built-in node that automatically executes tool calls from a language model with tools attached via bind_tools(), then sends the results back to the state as messages.

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool


@tool
def get_order_status(order_id: str) -> str:
    """Look up the status of an order."""
    return f"Order {order_id} is out for delivery."


tools = [get_order_status]


# Attach tools to the model
llm_with_tools = llm.bind_tools(tools)


# ToolNode executes any tool calls from the model
builder.add_node("tools", ToolNode(tools))

Learn more on Codecademy

  • Skill path

    Learn LangChain

    Build powerful AI applications using LangChain and LangGraph.
    • Includes 3 Courses
    • With Certificate
    • Intermediate.
      4 hours