LangGraph Essentials for Developers

Complete course on LangGraph — from concept to implementation, including agents, memory, streaming, and evaluation with LangSmith.

Table of Contents

1. LangGraph Fundamentals
   • Introduction and Core Components
   • Steps to Build a Workflow
   • Workflow Visualization
   • The State
   • State Evolution
   • Reducers and Message Accumulation
2. Flow Control Patterns
   • Sequential Execution
   • Parallel Execution
   • Reducers in Parallel Execution
   • Conditional Execution (Routing)
   • Iterative Execution (Loops)
   • Iterative Workflows with Feedback Loop
3. Tool Integration and ReAct Agents
   • Building a Basic Chatbot
   • Tools in LangGraph
   • Integrating Tools into a Workflow
   • Custom ReAct Agent
   • Built-in ReAct Agent (create_agent)
   • Multi-Agent System
4. Memory and Persistence
   • Understanding Persistence
   • Super-steps and Checkpoints
   • Separating Executions with Threads
   • InMemorySaver
   • SQLiteSaver
   • Short-Term Memory for a Chatbot
   • Fault Tolerance
5. Time Travel and Human-in-the-Loop
   • Understanding Time Travel
   • Implementing Time Travel
   • HITL — Concepts and Use Cases
   • Implementing HITL in LangGraph
   • Human Approval with Feedback Loop
   • Routing with Command
   • Interruptions in Tools
   • Static Interruptions for Debugging
6. Streaming and UI Integration
   • Streaming Basics
   • Streaming in Action
   • Streaming LLM Responses
   • Web Interface with Streamlit
7. Observability and Evaluation with LangSmith
   • Getting Started with LangSmith
   • Workflow Tracing
   • Debugging and Performance Analytics
   • Production Monitoring
   • Offline Evaluation
   • Creating Datasets
   • Running Experiments

Module 1 — LangGraph Fundamentals

Introduction and Core Components

LangGraph is an orchestration framework that structures and controls execution flow in agentic applications. It is developed by the LangChain team and integrates naturally with LangChain abstractions (models, prompts, tools), but can also be used standalone.

LangGraph is built on graph-based workflows:

graph LR
    START([__start__]) --> nodeA[Node A]
    nodeA --> nodeB[Node B]
    nodeB --> nodeC[Node C]
    nodeC --> END([__end__])

Core Components

Steps to Build a Workflow

The process of building a LangGraph application unfolds in 3 phases:

flowchart TD
    A["1. Define the State (TypedDict / Pydantic)"] --> B["2. Define the nodes (Python functions)"]
    B --> C["3. Create a StateGraph instance"]
    C --> D["4. Add nodes and edges"]
    D --> E["5. Compile the graph (.compile())"]
    E --> F["6. Execute the graph (.invoke(initial_state))"]

First Workflow — Email Processing

# LangGraph Example: Simple Email Processing Workflow

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

class EmailState(TypedDict):
    email_content: str
    is_spam: bool
    classification: str
    response: str

def check_spam(state: EmailState):
    spam_words = ["free", "winner", "urgent", "click now"]
    is_spam = any(word in state["email_content"].lower() for word in spam_words)
    print("check_spam node executed.")
    return {"is_spam": is_spam}

def classify_email(state: EmailState):
    if state["is_spam"]:
        classification = "spam"
    elif "support" in state["email_content"].lower():
        classification = "support"
    else:
        classification = "general"
    print("classify_email node executed.")
    return {"classification": classification}

def generate_response(state: EmailState):
    if state["classification"] == "spam":
        response = "Email moved to spam folder"
    elif state["classification"] == "support":
        response = "Forwarded to support team"
    else:
        response = "Email filed in inbox"
    print("generate_response node executed.")
    return {"response": response}

graph_builder = StateGraph(EmailState)
graph_builder.add_node("check_spam", check_spam)
graph_builder.add_node("classify_email", classify_email)
graph_builder.add_node("generate_response", generate_response)

graph_builder.add_edge(START, "check_spam")
graph_builder.add_edge("check_spam", "classify_email")
graph_builder.add_edge("classify_email", "generate_response")
graph_builder.add_edge("generate_response", END)

graph = graph_builder.compile()
initial_state = {"email_content": "Click now to get free gifts"}
final_state = graph.invoke(initial_state)
print(final_state)

graph LR
    START([__start__]) --> check_spam
    check_spam --> classify_email
    classify_email --> generate_response
    generate_response --> END([__end__])

Workflow Visualization

To generate a Mermaid PNG diagram of the workflow:

# In a Python script
with open("workflow.png", "wb") as f:
    f.write(graph.get_graph().draw_mermaid_png())

# In a Jupyter Notebook
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

The State

The State is the central data structure shared by all nodes. It can be defined in three ways:

1. TypedDict (most common)

from typing import TypedDict, Annotated, Literal

class TicketState(TypedDict):
    description: str | None          # optional value
    priority: int | str              # multiple types
    status: Literal["open", "closed", "pending"]  # predefined values
    tags: list[str]                  # list of strings
    intermediate_steps: list[tuple[str, str]]  # list of tuples
    counter: Annotated[int, "Iteration counter"]  # with metadata

2. Pydantic (runtime validation)

from pydantic import BaseModel, Field

class OrderState(BaseModel):
    order_id: str
    items: list[str] = []
    total: float = Field(default=0.0, ge=0)
    status: str = "pending"

3. Dataclass

from dataclasses import dataclass, field

@dataclass
class WorkflowState:
    topic: str = ""
    result: str = ""
    attempts: int = 0

Good to know: TypedDict does not perform runtime validation. Pydantic is slower but offers full validation with clear error messages.

State Evolution

LangGraph maintains an immutable state: at each step, it creates a new state object rather than modifying the existing one. This enables:

• A complete history of states (short-term memory)
• Debugging and rollback
• Time Travel (see Module 5)

sequenceDiagram
    participant G as Graph
    participant S0 as State₀ (initial)
    participant S1 as State₁ (after node 1)
    participant S2 as State₂ (after node 2)
    participant S3 as State₃ (final)
    
    G->>S0: invoke(initial_state)
    S0->>S1: check_spam → {is_spam: true}
    S1->>S2: classify_email → {classification: "spam"}
    S2->>S3: generate_response → {response: "..."}
    G->>G: returns S3

Important rule: Nodes must never mutate the state directly. They must always return a dictionary containing only the modified keys.

# ❌ Avoid
def bad_node(state):
    state["key"] = "value"  # direct mutation
    return state

# ✅ Correct
def good_node(state):
    return {"key": "value"}  # return only the updates

Reducers and Message Accumulation

By default, if multiple nodes update the same key, the new value overwrites the old one. Reducers define how to merge these updates.

Problem Without a Reducer

# Without reducer, only the last node's log is kept
class EmailState(TypedDict):
    log: list[str]  # the 3rd node will overwrite the first 2

Solution With a Reducer

from typing import Annotated

def log_reducer(existing, new):
    return existing + new

class EmailState(TypedDict):
    email_content: str
    is_spam: bool
    classification: str
    response: str
    log: Annotated[list[str], log_reducer]  # logs accumulate

Common Reducers

import operator
from typing import Annotated

class State(TypedDict):
    # Add numbers
    total: Annotated[float, operator.add]
    
    # Keep the minimum value
    min_val: Annotated[float, lambda a, b: min(a, b)]
    
    # Keep the maximum value
    max_val: Annotated[float, lambda a, b: max(a, b)]
    
    # Merge dictionaries
    metadata: Annotated[dict, lambda a, b: {**a, **b}]

add_messages — The Reducer for Chatbots

add_messages is a specialized reducer for chat-style workflows. It:

• Understands LangChain message types (AIMessage, HumanMessage, etc.)
• Handles duplicate messages automatically
• Converts raw data to appropriate message objects

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def node1(state: ChatState):
    return {"messages": [HumanMessage(content="Hello there!")]}

def node2(state: ChatState):
    return {"messages": [AIMessage(content="Hi! How can I assist you today?")]}

def node3(state: ChatState):
    return {"messages": ["Hello"]}  # string auto-converted

def node4(state: ChatState):
    return {"messages": [{"type": "human", "content": "What can you do?"}]}  # dict auto-converted

graph_builder = StateGraph(ChatState)
# ... add nodes and edges ...
graph = graph_builder.compile()

final_state = graph.invoke({"messages": []})
for message in final_state["messages"]:
    message.pretty_print()

Module 2 — Flow Control Patterns

LangGraph supports 4 main execution patterns:

graph TB
    subgraph "Execution Patterns"
        A["Sequential\n(one after another)"]
        B["Parallel\n(simultaneously)"]
        C["Conditional\n(state-based routing)"]
        D["Iterative\n(loops)"]
    end

Sequential Execution

Tasks execute one after another in a fixed order. Ideal for step-by-step processing pipelines.

graph LR
    START([__start__]) --> generate_passage
    generate_passage --> extract_key_points
    extract_key_points --> generate_questions
    generate_questions --> END([__end__])

# Sequential pipeline: topic → passage → key points → questions
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

class WorkflowState(TypedDict):
    topic: str
    passage: str
    key_points: str
    questions: str

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

def generate_passage(state: WorkflowState):
    response = llm.invoke(f"Write a short informative passage about: {state['topic']}")
    return {"passage": response.content}

def extract_key_points(state: WorkflowState):
    response = llm.invoke(f"Extract 5 key points from this passage:\n\n{state['passage']}")
    return {"key_points": response.content}

def generate_questions(state: WorkflowState):
    response = llm.invoke(f"Based on these key points, generate 5 quiz questions:\n\n{state['key_points']}")
    return {"questions": response.content}

graph_builder = StateGraph(WorkflowState)
graph_builder.add_node("generate_passage", generate_passage)
graph_builder.add_node("extract_key_points", extract_key_points)
graph_builder.add_node("generate_questions", generate_questions)

graph_builder.add_edge(START, "generate_passage")
graph_builder.add_edge("generate_passage", "extract_key_points")
graph_builder.add_edge("extract_key_points", "generate_questions")
graph_builder.add_edge("generate_questions", END)

graph = graph_builder.compile()
final_state = graph.invoke({"topic": "Artificial Intelligence in Education"})

Parallel Execution

Independent tasks execute simultaneously — the fan-out / fan-in technique:

graph TD
    START([__start__]) --> strengths["extract_strengths"]
    START --> weaknesses["extract_weaknesses"]
    START --> questions["generate_interview_questions"]
    strengths --> merge_report["create_screening_report"]
    weaknesses --> merge_report
    questions --> merge_report
    merge_report --> END([__end__])

# Parallel workflow: candidate screening analysis
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict
from dotenv import load_dotenv

load_dotenv()

class ScreeningState(TypedDict):
    cv_text: str
    job_description: str
    strengths: str
    weaknesses: str
    interview_questions: str
    screening_report: str

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

def extract_strengths(state: ScreeningState):
    response = llm.invoke(f"""
    CV: {state["cv_text"]}
    Job Description: {state["job_description"]}
    Extract the candidate's key strengths.
    """)
    return {"strengths": response.content}

def extract_weaknesses(state: ScreeningState):
    response = llm.invoke(f"""
    CV: {state["cv_text"]}
    Job Description: {state["job_description"]}
    Extract the candidate's weaknesses or skill gaps.
    """)
    return {"weaknesses": response.content}

def generate_interview_questions(state: ScreeningState):
    response = llm.invoke(f"""
    CV: {state["cv_text"]}
    Job Description: {state["job_description"]}
    Suggest 5 tailored interview questions.
    """)
    return {"interview_questions": response.content}

def create_screening_report(state: ScreeningState):
    response = llm.invoke(f"""
    Create a structured screening report:
    Strengths: {state["strengths"]}
    Weaknesses: {state["weaknesses"]}
    Interview Questions: {state["interview_questions"]}
    """)
    return {"screening_report": response.content}

graph_builder = StateGraph(ScreeningState)
graph_builder.add_node("strengths", extract_strengths)
graph_builder.add_node("weaknesses", extract_weaknesses)
graph_builder.add_node("questions", generate_interview_questions)
graph_builder.add_node("merge_report", create_screening_report)

# Fan-out: START to 3 nodes in parallel
graph_builder.add_edge(START, "strengths")
graph_builder.add_edge(START, "weaknesses")
graph_builder.add_edge(START, "questions")

# Fan-in: 3 nodes converge into merge_report
graph_builder.add_edge("strengths", "merge_report")
graph_builder.add_edge("weaknesses", "merge_report")
graph_builder.add_edge("questions", "merge_report")
graph_builder.add_edge("merge_report", END)

graph = graph_builder.compile()

Reducers in Parallel Execution

When multiple parallel nodes update the same key, a reducer is mandatory:

from pydantic import BaseModel, Field
from typing import Annotated

# Structured output schema for the LLM
class StrengthsOutput(BaseModel):
    strengths_text: str
    strength_score: float = Field(ge=0, le=1)  # between 0 and 1

class WeaknessesOutput(BaseModel):
    weaknesses_text: str
    weakness_score: float = Field(ge=-1, le=0)  # between -1 and 0

class ScreeningState(TypedDict):
    cv_text: str
    job_description: str
    strengths: str
    weaknesses: str
    interview_questions: str
    screening_report: str
    # Reducer mandatory because updated by 2 parallel nodes
    candidate_score: Annotated[float, operator.add]

# LLMs with structured output
strengths_llm = llm.with_structured_output(StrengthsOutput)
weaknesses_llm = llm.with_structured_output(WeaknessesOutput)

Rule: Without a reducer on a key updated by multiple parallel nodes, LangGraph raises an error.

Conditional Execution (Routing)

The conditional edge allows routing the flow to different nodes based on state:

graph TD
    START([__start__]) --> preprocess
    preprocess --> categorize
    categorize -->|billing| billing_node
    categorize -->|technical| tech_node
    categorize -->|general| clarify_node
    billing_node --> summarize
    tech_node --> summarize
    clarify_node --> summarize
    summarize --> END([__end__])

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
from dotenv import load_dotenv

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

class SupportState(TypedDict):
    email_text: str
    cleaned_text: str
    category: str
    response: str
    summary: str

def categorize_email(state: SupportState):
    prompt = f"""Classify this email as: billing, technical, or general.
Email: {state["cleaned_text"]}
Respond with only one word."""
    predicted = llm.invoke(prompt).content.strip().lower()
    if predicted not in {"billing", "technical", "general"}:
        predicted = "general"
    return {"category": predicted}

# Routing function — returns the name of the next node
def route_next(state: SupportState) -> Literal["billing_node", "tech_node", "clarify_node"]:
    c = state["category"]
    if "bill" in c:
        return "billing_node"
    elif "tech" in c:
        return "tech_node"
    else:
        return "clarify_node"

graph_builder = StateGraph(SupportState)
# ... add nodes ...
graph_builder.add_conditional_edges("categorize", route_next)

Syntax with label mapping:

graph_builder.add_conditional_edges(
    "source_node",
    routing_function,
    {
        "label_a": "node_a",
        "label_b": "node_b",
        "label_c": END
    }
)

Iterative Execution (Loops)

Loops are implemented by redirecting flow back to an already-executed node via a conditional edge:

graph TD
    START([__start__]) --> task_node
    task_node -->|"remaining tasks"| task_node
    task_node -->|"empty tasks"| END([__end__])

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

class State(TypedDict):
    tasks: list[str]

def task_node(state: State):
    task = state["tasks"][0]
    print(f"Processing task: {task}")
    return {"tasks": state["tasks"][1:]}  # removes the first task

def should_continue(state: State) -> Literal["loop", "exit"]:
    if state["tasks"]:
        return "loop"    # return to the same node → creates the loop
    else:
        return "exit"    # exit

graph_builder = StateGraph(State)
graph_builder.add_node("task_node", task_node)
graph_builder.add_conditional_edges(
    "task_node",
    should_continue,
    {
        "loop": "task_node",   # points to the same node
        "exit": END
    }
)
graph_builder.add_edge(START, "task_node")
graph = graph_builder.compile()

graph.invoke({"tasks": ["Email client", "Write report", "Schedule meeting"]})

Iterative Workflows with Feedback Loop

The evaluator-optimizer pattern uses a loop to iteratively improve an output:

graph TD
    START([__start__]) --> post_generator
    post_generator --> post_evaluator
    post_evaluator -->|"ready to post\nor max 3 attempts"| END([__end__])
    post_evaluator -->|"needs rewrite"| post_generator

from typing import TypedDict, Literal
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

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

class State(TypedDict):
    topic: str
    post: str
    feedback: str
    quality: Literal["ready to post", "needs rewrite"]
    attempts: int

class Review(BaseModel):
    verdict: Literal["ready to post", "needs rewrite"] = Field(
        description="Decide if the LinkedIn post is ready or needs improvement."
    )
    feedback: str = Field(description="Suggest how to improve if rewrite is needed.")

evaluator_llm = llm.with_structured_output(Review)

def post_generator(state: State):
    if state.get("feedback") and state.get("post"):
        prompt = (f"Previous post:\n{state['post']}\n\n"
                  f"Feedback: {state['feedback']}\n\n"
                  f"Revise the post about '{state['topic']}'.")
    else:
        prompt = f"Write an engaging LinkedIn post about '{state['topic']}'."
    response = llm.invoke(prompt)
    return {"post": response.content}

def post_evaluator(state: State):
    review = evaluator_llm.invoke(f"Evaluate this LinkedIn post:\n{state['post']}")
    return {
        "quality": review.verdict,
        "feedback": review.feedback,
        "attempts": state.get("attempts", 0) + 1
    }

def decide_next(state: State):
    if state["quality"] == "ready to post" or state["attempts"] >= 3:
        return "Accept"   # max 3 attempts to avoid infinite loop
    return "Revise"

graph_builder = StateGraph(State)
graph_builder.add_node("post_generator", post_generator)
graph_builder.add_node("post_evaluator", post_evaluator)
graph_builder.add_edge(START, "post_generator")
graph_builder.add_edge("post_generator", "post_evaluator")
graph_builder.add_conditional_edges("post_evaluator", decide_next, {
    "Accept": END,
    "Revise": "post_generator"
})
graph = graph_builder.compile()
final_state = graph.invoke({"topic": "Personal Branding for Creators", "attempts": 0})
print("Final LinkedIn Post:\n", final_state["post"])

Important: Always add an attempt counter (attempts) to avoid infinite loops.

Module 3 — Tool Integration and ReAct Agents

Building a Basic Chatbot

A simple LangGraph chatbot contains a single node with an LLM:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from dotenv import load_dotenv

load_dotenv()

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

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

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

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

# Multi-turn conversation with Python loop
while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "bye"]:
        break
    final_state = graph.invoke({"messages": [HumanMessage(content=user_input)]})
    print("Assistant:", final_state["messages"][-1].content)

Tools in LangGraph

Tools provide access to external functionality (web search, computations, APIs):

sequenceDiagram
    participant U as User
    participant LLM as LLM
    participant T as Tool (executed by framework)
    
    U->>LLM: "What is the weather in Toronto?"
    LLM->>LLM: Decides a tool is needed
    LLM-->>T: tool_call: {name: "search", args: {query: "Toronto weather"}}
    T-->>LLM: Search result
    LLM->>U: "It's 22°C in Toronto today."

Types of tools:

from langchain_core.tools import tool
from langchain_tavily import TavilySearch

# Built-in tool
tavily_tool = TavilySearch(max_results=2)

# Custom tool with the @tool decorator
@tool
def add(a: int, b: int) -> int:
    """Adds two numbers and returns the sum."""
    return a + b

@tool("subtract", description="Subtract b from a and return the result.")
def sub(a: int, b: int) -> int:
    """Return the difference of two numbers."""
    return a - b

# Bind tools to the LLM
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools([tavily_tool, add, sub])

Integrating Tools into a Workflow

graph LR
    START([__start__]) --> chatbot
    chatbot -->|"tool_call"| tool_node
    chatbot -->|"text response"| END([__end__])
    tool_node --> END

from typing import TypedDict, Annotated
from langgraph.graph import add_messages, StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage, ToolMessage
from langchain_core.tools import tool
from langchain_tavily import TavilySearch
from dotenv import load_dotenv

load_dotenv()

tavily_tool = TavilySearch(max_results=2)

@tool
def add(a: int, b: int) -> int:
    """Adds two numbers and returns the sum."""
    return a + b

llm = ChatOpenAI(model="gpt-4o-mini")
tools = [tavily_tool, add]
llm_with_tools = llm.bind_tools(tools)
tools_by_name = {tool.name: tool for tool in tools}

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def chatbot(state: ChatState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": response}

def should_continue(state: ChatState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "invoke tools"
    return "end"

def tool_node(state: ChatState):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        tool_output = tool.invoke(tool_call["args"])
        result.append(ToolMessage(
            content=tool_output,
            tool_call_id=tool_call["id"]
        ))
    return {"messages": result}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tool_node", tool_node)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges(
    "chatbot", should_continue,
    {"invoke tools": "tool_node", "end": END}
)
graph_builder.add_edge("tool_node", END)
graph = graph_builder.compile()

Custom ReAct Agent

The ReAct (Reason + Act) agent creates a loop between the chatbot and the tool node, enabling sequential tool calls and natural language responses:

graph LR
    START([__start__]) --> chatbot
    chatbot -->|"tool_call"| tool_node
    chatbot -->|"final response"| END([__end__])
    tool_node --> chatbot

Only one line changes from the previous workflow:

# BEFORE (chatbot without tool memory)
graph_builder.add_edge("tool_node", END)

# AFTER (ReAct: return to chatbot to reason over the result)
graph_builder.add_edge("tool_node", "chatbot")

The Reason → Act → Observe → Reason cycle repeats until the LLM can answer in natural language.

Built-in ReAct Agent (create_agent)

Instead of writing all the boilerplate code, you can use create_agent:

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
from dotenv import load_dotenv

load_dotenv()

tavily_tool = TavilySearch(max_results=2)

@tool
def add(a: int, b: int) -> int:
    """Adds two numbers and returns the sum."""
    return a + b

@tool("subtract", description="Subtract b from a and return the result.")
def sub(a: int, b: int) -> int:
    """Return a - b."""
    return a - b

llm = ChatOpenAI(model="gpt-4o-mini")
tools = [tavily_tool, add, sub]

agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="You are a helpful assistant. Use tools for web search or math calculations."
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "Find temperature of Bangalore and then add 5 to it"}]
})

# Display all messages
for msg in result["messages"]:
    msg.pretty_print()

# Display only the final response
print("\nAssistant:", result["messages"][-1].content)

Multi-Agent System

For complex tasks, a multi-agent architecture is used with a controller agent that delegates to sub-agents:

graph TD
    U[User] --> controller["Controller Agent"]
    controller -->|"math question"| math_wrapper["math_helper (tool)"]
    controller -->|"research question"| search_wrapper["search_helper (tool)"]
    math_wrapper --> math_agent["Math Agent\n(add, subtract)"]
    search_wrapper --> research_agent["Research Agent\n(TavilySearch)"]
    math_agent --> controller
    research_agent --> controller
    controller --> U

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
from dotenv import load_dotenv

load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Math sub-agent
@tool(description="Add two numbers.")
def add(a: int, b: int) -> int:
    return a + b

@tool("subtract", description="Subtract b from a.")
def sub(a: int, b: int) -> int:
    return a - b

math_agent = create_agent(model=llm, tools=[add, sub],
    system_prompt="You are a math expert. Always use tools for calculations.")

@tool("math_helper", description="Use this for arithmetic questions.")
def call_math_agent(query: str) -> str:
    result = math_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# Research sub-agent
tavily_tool = TavilySearch(max_results=2)
research_agent = create_agent(model=llm, tools=[tavily_tool],
    system_prompt="You are a research specialist. Use TavilySearch for information.")

@tool("search_helper", description="Use this to retrieve up-to-date information.")
def call_research_agent(query: str) -> str:
    result = research_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# Controller agent
controller_agent = create_agent(
    model=llm,
    tools=[call_math_agent, call_research_agent],
    system_prompt="You are a controller. Decide which agent should perform the task."
)

queries = [
    "What is 56 - 19?",
    "What is the young one of a cat called?",
    "Add 5 to the temperature in Bangalore today.",
]

for q in queries:
    print(f"\nUser: {q}")
    response = controller_agent.invoke({"messages": [{"role": "user", "content": q}]})
    print("Assistant:", response["messages"][-1].content)

Module 4 — Memory and Persistence

Understanding Persistence

By default, a workflow’s state is ephemeral: it disappears at the end of execution. Persistence allows saving and reloading the state.

Capabilities enabled by persistence:

Super-steps and Checkpoints

graph LR
    subgraph "Super-step 1"
        A["Node A"]
    end
    subgraph "Super-step 2 (parallel)"
        B["Node B"]
        C["Node C"]
    end
    subgraph "Super-step 3"
        D["Node D"]
    end
    START --> A --> B & C --> D --> END

• Super-step: A set of nodes that execute together (sequential nodes = different super-steps, parallel nodes = same super-step)
• Checkpoint: State snapshot saved at the end of each super-step

A checkpoint (StateSnapshot object) contains:

• values: the current workflow state
• next: the next node to execute
• metadata: additional information (step, timestamps, etc.)

Separating Executions with Threads

Each independent workflow execution receives a unique thread_id. All checkpoints of the same execution are grouped under this thread_id.

# Pass thread_id when invoking
config = {"configurable": {"thread_id": "user-1"}}
final_state = graph.invoke(initial_state, config=config)

# Re-invoke with the same thread_id → resumes from the last checkpoint
final_state = graph.invoke(new_input, config=config)

InMemorySaver

Stores checkpoints in RAM — ideal for tests and development:

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    data: Annotated[str, operator.add]

# ... node definitions ...

checkpointer = InMemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "demo-thread-1"}}
final_state = graph.invoke({"data": ""}, config=config)

# Get the latest checkpoint
latest = graph.get_state(config)
print(latest.values)   # current state
print(latest.next)     # next node (empty if finished)

# Get the full checkpoint history
for checkpoint in graph.get_state_history(config):
    print(checkpoint.metadata["step"], checkpoint.values, checkpoint.next)

Limitation: Data is lost if the program stops.

SQLiteSaver

Stores checkpoints in a SQLite database on disk — for production:

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# Connect to the SQLite database
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)

graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-1"}}
final_state = graph.invoke(initial_state, config=config)
# Checkpoints are now saved to disk

Short-Term Memory for a Chatbot

Using SQLite persistence, the chatbot remembers all past conversations:

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# Same ReAct workflow as before, but compiled with a checkpointer
conn = sqlite3.connect("chatbot_memory.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=checkpointer)

# Fixed thread for the entire conversation session
config = {"configurable": {"thread_id": "session-001"}}

while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "bye"]:
        break
    # LangGraph automatically loads history for this thread_id
    final_state = graph.invoke(
        {"messages": [HumanMessage(content=user_input)]},
        config=config
    )
    print("Assistant:", final_state["messages"][-1].content)

Fault Tolerance

In case of a crash, you can resume from the last checkpoint:

# Health monitoring workflow with SQLite
from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

class HealthState(TypedDict):
    patient_name: str
    risk_score: Annotated[float, operator.add]  # accumulate scores
    health_status: str

# ... node definitions ...

conn = sqlite3.connect("health_checkpoints.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "patient-001"}}

# First execution (may crash halfway)
try:
    final_state = graph.invoke(
        {"patient_name": "Tim Dave", "risk_score": 0.0},
        config=config
    )
except KeyboardInterrupt:
    print("Simulated crash! Checkpoints are saved.")

# Resume after crash: pass None instead of the initial state
# LangGraph automatically loads the last checkpoint
final_state = graph.invoke(None, config=config)
print(f"Health Status: {final_state['health_status']}")

Module 5 — Time Travel and Human-in-the-Loop

Understanding Time Travel

Time Travel allows re-executing a workflow from a previous checkpoint:

timeline
    title Checkpoint History
    Checkpoint 1 : validate_order → State₁
    Checkpoint 2 : check_inventory → State₂
    Checkpoint 3 : fulfill_order → State₃
    Checkpoint 4 : generate_message → State₄ (final)

• Original checkpoints remain intact
• LangGraph creates a new branch in the history
• Use cases: debugging, “what-if” scenarios, exploring alternative paths

Implementing Time Travel

# 1. Execute the workflow normally
config = {"configurable": {"thread_id": "order-thread-1"}}
final_state = graph.invoke(initial_order_state, config=config)

# 2. List all checkpoints
checkpoints = list(graph.get_state_history(config))
for cp in checkpoints:
    print(cp.config["configurable"]["checkpoint_id"], "→ next:", cp.next)

# 3. Choose a checkpoint for time travel
target_checkpoint_id = checkpoints[2].config["configurable"]["checkpoint_id"]

# 4. Re-execute from this checkpoint
time_travel_config = {
    "configurable": {
        "thread_id": "order-thread-1",
        "checkpoint_id": target_checkpoint_id  # specify the exact checkpoint
    }
}
new_state = graph.invoke(None, config=time_travel_config)

HITL — Concepts and Use Cases

HITL (Human-in-the-Loop) integrates human judgment into automated workflows:

graph TD
    A["LLM Workflow"] -->|"risky action"| B{{"Pause — Awaiting human"}}
    B -->|"Approve"| C["Execute action"]
    B -->|"Reject"| D["Alternative path"]
    B -->|"Modify + feedback"| E["Improve and retry"]
    C --> F["Continue workflow"]
    D --> F
    E --> A

Common HITL patterns:

Why Python input() is not enough:

• Blocks the entire program
• Does not support multi-user workflows
• Does not survive restarts

Implementing HITL in LangGraph

LangGraph uses the interrupt() function and the Command class:

sequenceDiagram
    participant W as Workflow
    participant H as Human (UI)
    
    W->>W: Normal execution
    W->>W: interrupt() called
    W-->>H: Payload (data to review)
    Note over W: Pause — state saved
    H->>W: graph.invoke(Command(resume=response))
    W->>W: Resumes execution

from langgraph.types import interrupt, Command

def human_review_node(state):
    # Pause and send a payload to the interface
    human_input = interrupt({
        "tweet": state["messages"][-1].content,
        "question": "Do you want to post this tweet? (yes/no)"
    })
    return {"approval": human_input}

Simple Human Approval

# Tweet approval workflow
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, add_messages, START, END
from langgraph.types import interrupt, Command
from langchain_core.messages import HumanMessage, SystemMessage, AnyMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from dotenv import load_dotenv

load_dotenv()

class State(TypedDict):
    topic: str
    messages: Annotated[list[AnyMessage], add_messages]
    approval: str

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

def create_tweet(state: State):
    response = llm.invoke([
        SystemMessage("You are an expert at writing engaging tweets"),
        HumanMessage(f"Write a tweet on {state['topic']}")
    ])
    return {"messages": response}

def human_review(state: State):
    human_input = interrupt({
        "tweet": state["messages"][-1].content,
        "question": "Do you want to post this tweet? (yes/no)"
    })
    return {"approval": human_input}

def review_router(state: State) -> Literal["post_tweet", "__end__"]:
    return "post_tweet" if state["approval"].lower() == "yes" else END

def post_tweet(state: State):
    print("\nTweet posted:\n", state["messages"][-1].content)
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("create_tweet", create_tweet)
graph_builder.add_node("human_review", human_review)
graph_builder.add_node("post_tweet", post_tweet)
graph_builder.add_edge(START, "create_tweet")
graph_builder.add_edge("create_tweet", "human_review")
graph_builder.add_conditional_edges("human_review", review_router)
graph_builder.add_edge("post_tweet", END)

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

config = {"configurable": {"thread_id": "demo-1"}}
result = graph.invoke({"topic": "Life"}, config=config)

# Extract and display the payload
payload = result["__interrupt__"][0].value
print("Generated tweet:", payload["tweet"])

# Simulate human response (would come from a UI in production)
human_response = "yes"
graph.invoke(Command(resume=human_response), config=config)

Human Approval with Feedback Loop

graph TD
    START --> create_tweet
    create_tweet --> human_review
    human_review -->|"yes"| post_tweet
    human_review -->|"no + feedback"| incorporate_feedback
    incorporate_feedback --> human_review
    post_tweet --> END([__end__])

def human_review(state: State):
    human_input = interrupt({
        "tweet": state["messages"][-1].content,
        "question": "Approve? yes/no + feedback if no"
    })
    # human_input expected: {"approve": "yes"} or {"approve": "no", "feedback": "..."}
    return {
        "approval": human_input.get("approve", "no"),
        "feedback": human_input.get("feedback", "")
    }

def incorporate_feedback(state: State):
    response = llm.invoke([
        SystemMessage("Rewrite the tweet using the feedback."),
        HumanMessage(f"Tweet: {state['messages'][-1].content}\nFeedback: {state['feedback']}")
    ])
    return {"messages": response}

Routing with Command

The Command class replaces conditional edges — the node itself decides the next node:

from langgraph.types import Command
from typing import Literal

# With conditional edge (classic approach)
def classify_and_route_classic(state):
    category = "sales" if "price" in state["email_text"] else "support"
    return {"category": category}  # routing is in a separate function

# With Command (everything in the same node)
def classify_and_route(state) -> Command[Literal["handle_sales", "handle_support"]]:
    text = state["email_text"].lower()
    if "price" in text or "buy" in text:
        return Command(goto="handle_sales")
    else:
        return Command(goto="handle_support")

# Update state AND route at the same time
def classify_route_and_update(state) -> Command[Literal["handle_sales", "handle_support"]]:
    if "price" in state["email_text"]:
        return Command(
            goto="handle_sales",
            update={"assigned_team": "Sales"}  # state update
        )
    else:
        return Command(
            goto="handle_support",
            update={"assigned_team": "Support"}
        )

Interruptions in Tools

Interruptions can be placed directly inside a tool — the tool becomes reusable with built-in HITL:

from langchain_core.tools import tool
from langgraph.types import interrupt

@tool
def delete_file(file_name: str):
    """Delete a file from the system (simulated)."""
    human_input = interrupt({
        "action": "delete_file",
        "file_name": file_name,
        "message": "Do you really want to delete this file?"
    })
    if human_input.get("approve") == "yes":
        return f"File '{file_name}' deleted successfully."
    return "File deletion cancelled by user."

# This tool can be provided to any ReAct workflow
# HITL approval is automatic on every call
llm_with_tools = llm.bind_tools([delete_file])

Advantage: Any workflow using this tool automatically benefits from human approval, without additional logic in the workflow.

Static Interruptions for Debugging

Static interruptions work like breakpoints — for debugging only:

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["B"],         # pause before node B
    interrupt_after=["C", "D"]      # pause after nodes C and D
)

config = {"configurable": {"thread_id": "debug-1"}}

# First invocation: stops before B
graph.invoke({"data": ""}, config=config)
# Inspect state here...

# Continue: stops after C
graph.invoke(None, config=config)

# Continue: stops after D
graph.invoke(None, config=config)

# Continue to the end
graph.invoke(None, config=config)

Module 6 — Streaming and UI Integration

Streaming Basics

invoke vs stream:

Streaming modes:

# "values" mode: full state after each step
for chunk in graph.stream(initial_state, mode="values"):
    print(chunk)

# "updates" mode: only what changed
for chunk in graph.stream(initial_state, mode="updates"):
    # chunk = {node_name: {modified_keys}}
    print(chunk)

# "debug" mode: all execution information
for chunk in graph.stream(initial_state, mode="debug"):
    print(chunk)

Streaming in Action

# Email support workflow with streaming (updates mode)
for chunk in graph.stream({"email_text": email}, mode="updates"):
    node_name = list(chunk.keys())[0]
    state_update = list(chunk.values())[0]
    print(f"[{node_name}] → {state_update}")

# Debug mode with readable formatting
for chunk in graph.stream(initial_state, mode="debug"):
    if "step" in chunk:
        print(f"Step {chunk['step']} | Node: {chunk.get('node')} | Updates: {chunk.get('updates')}")

Advantage of streaming with loops: In iterative workflows, each iteration is visible in real time.

Streaming LLM Responses

Use the messages mode for a live typing effect:

from langchain_core.messages import AIMessageChunk

# ReAct workflow with LLM token streaming
while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "bye"]:
        break

    print("Assistant: ", end="", flush=True)
    for chunk, metadata in graph.stream(
        {"messages": [HumanMessage(content=user_input)]},
        config=config,
        stream_mode="messages"
    ):
        # Filter only model chunks (not ToolMessages)
        if isinstance(chunk, AIMessageChunk) and chunk.content:
            print(chunk.content, end="", flush=True)
    print()  # new line

Web Interface with Streamlit

LangGraph as the backend, Streamlit as the frontend:

graph LR
    U[User / Browser] <-->|"HTTP"| S[Streamlit Frontend\n(app.py)]
    S <-->|"Python import"| G[LangGraph Backend\n(graph_backend.py)]
    G <-->|"API calls"| LLM[OpenAI LLM]
    G <-->|"SQLite"| DB[(Checkpoints DB)]

graph_backend.py (unchanged — the LangGraph workflow):

# graph_backend.py — LangGraph Backend (tools, nodes, edges)
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# ... ReAct workflow definition ...

conn = sqlite3.connect("chatbot.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "streamlit-session-1"}}

app.py (Streamlit interface):

import streamlit as st
from langchain_core.messages import HumanMessage
from graph_backend import graph, config

st.set_page_config(page_title="LangGraph Chatbot", page_icon="🤖")
st.title("🤖 LangGraph Chatbot")
st.caption("A Streamlit UI on top of a LangGraph chatbot with tools and memory")

# Conversation history in Streamlit session
if "messages" not in st.session_state:
    st.session_state.messages = []

# Display previous messages
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# Input area
user_input = st.chat_input("Ask me anything...")

if user_input:
    st.session_state.messages.append({"role": "user", "content": user_input})
    with st.chat_message("user"):
        st.markdown(user_input)

    with st.spinner("Thinking..."):
        final_state = graph.invoke(
            {"messages": [HumanMessage(content=user_input)]},
            config=config
        )

    assistant_reply = final_state["messages"][-1].content
    st.session_state.messages.append({"role": "assistant", "content": assistant_reply})
    with st.chat_message("assistant"):
        st.markdown(assistant_reply)

Launch the application:

streamlit run app.py

Module 7 — Observability and Evaluation with LangSmith

Getting Started with LangSmith

LangSmith is an observability and evaluation platform specialized for LLM applications. It automatically traces every workflow execution.

Why LangSmith is needed:

• LLMs are non-deterministic — the same inputs don’t always produce the same outputs
• Failures are often semantic and subtle (no clear exception)
• Traditional logs don’t capture LLM inputs/outputs, latency, and costs

Configuration:

# Installation
pip install langsmith

# Environment variables (.env)
LANGSMITH_API_KEY=your_api_key_here
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-project-name

import os
from dotenv import load_dotenv

load_dotenv()
os.environ["LANGSMITH_PROJECT"] = "job-application-assistant"
# Tracing is automatic — no other code changes required!

Workflow Tracing

LangSmith hierarchy:

graph TD
    P["Project\n(application container)"]
    P --> T1["Trace 1\n(one complete execution)"]
    P --> T2["Trace 2"]
    T1 --> R1["Run: Node A"]
    T1 --> R2["Run: Node B (parallel)"]
    T1 --> R3["Run: Node C (parallel)"]
    R1 --> RR1["Sub-run: LLM call"]
    R2 --> RR2["Sub-run: Tool call"]

Naming traces:

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(
    run_name="Candidate screening - John Doe",  # trace name
    tags=["production", "v2"],
    metadata={"user_id": "123", "environment": "prod"}
)
final_state = graph.invoke(initial_state, config=config)

Debugging and Performance Analytics

Identifying failures:

# Simulate an error to see the trace in LangSmith
def merge_report(state: ScreeningState):
    raise RuntimeError("Simulated failure in merge_report node")
    # This error will be captured and visible in LangSmith
    # with the exact node, error message, and stack trace

In LangSmith:

• Failed traces are highlighted in red
• You can see exactly which run failed and the error message
• Error rate is displayed at the project level

Analyzing performance:

• Filter traces by latency, cost, tags, or metadata
• Compare traces side-by-side
• Identify bottlenecks (slow or costly nodes)

Production Monitoring

The monitoring dashboard shows trends over long periods:

┌─────────────────────────────────────────────────┐
│  Monitoring Dashboard                           │
├───────────────┬─────────────────────────────────┤
│ Volume        │ Number of traces / period       │
│ Latency       │ P50 (typical), P99 (worst case) │
│ Error rate    │ % of failed traces              │
│ LLM calls     │ Number of model calls           │
│ Tokens        │ Token consumption               │
│ Cost          │ Estimated cost in USD           │
└───────────────┴─────────────────────────────────┘

• Group by: tags, metadata, environment (advanced filtering)
• P50: median response time
• P99: response time of the slowest 1% (worst case)

Offline Evaluation

Offline evaluation tests the application before it reaches users:

graph LR
    D["Dataset\n(golden examples)"] --> E["Experiment\n(run app on each example)"]
    E --> EV["Evaluators\n(compare output vs reference)"]
    EV --> R["Results and scores"]

Types of evaluators:

Creating Datasets

A dataset is a collection of examples with inputs and reference outputs (the “golden examples”):

Methods for adding examples:

1. Manual: enter directly in the interface
2. From existing traces: select good executions
3. LLM generation: synthetically create from patterns
4. Programmatic: via the LangSmith API

Running Experiments

# run_eval.py — Evaluation script
from langsmith import evaluate
from job_screening import run_job_screening

# Evaluator 1: check that output exists
def has_output(outputs: dict, reference_outputs: dict):
    screening_report = outputs.get("screening_report", "")
    return 1 if screening_report.strip() else 0

# Evaluator 2: check that length is similar to reference
def similar_length(outputs: dict, reference_outputs: dict):
    actual = outputs.get("screening_report", "")
    expected = reference_outputs.get("screening_report", "")
    return 1 if abs(len(actual) - len(expected)) < 500 else 0

# Run the evaluation
evaluate(
    run_job_screening,        # application entry function
    data="candidate-dataset", # dataset name in LangSmith
    evaluators=[has_output, similar_length],
    experiment_prefix="job-screening-v1"
)

Entry function for LangSmith:

# job_screening.py — LangGraph application wrapped for LangSmith
def run_job_screening(inputs: dict) -> dict:
    """
    Entry point for LangSmith.
    LangSmith calls this function once per dataset example.
    """
    final_state = graph.invoke({
        "cv_text": inputs["cv_text"],
        "job_description": inputs["job_description"]
    })
    return {"screening_report": final_state["screening_report"]}

LLM-as-judge Evaluator (Correctness):

• Uses an LLM to compare output with reference
• Can be added directly from the LangSmith interface
• Applies automatically to all future experiments on this dataset
• Provides a semantic correctness score and justification

Summary — Complete Architecture

graph TB
    subgraph "Application Layer"
        UI["Streamlit / CLI / API"]
    end
    
    subgraph "LangGraph Core"
        G["Graph (StateGraph)"]
        S["State (TypedDict / Pydantic)"]
        N["Nodes (Python functions)"]
        E["Edges (normal + conditional)"]
        R["Reducers (state merging)"]
    end
    
    subgraph "Execution Patterns"
        SEQ["Sequential"]
        PAR["Parallel (fan-out/in)"]
        COND["Conditional (routing)"]
        ITER["Iterative (loops)"]
    end
    
    subgraph "Persistence"
        IMem["InMemorySaver (dev)"]
        SQLite["SQLiteSaver (prod)"]
        CP["Checkpoints"]
        TT["Time Travel"]
        FT["Fault Tolerance"]
    end
    
    subgraph "HITL"
        INT["interrupt()"]
        CMD["Command class"]
        TI["Tool Interrupts"]
        SI["Static Interrupts (debug)"]
    end
    
    subgraph "Agents"
        RC["ReAct Agent"]
        MA["Multi-Agent"]
        CA["create_agent()"]
    end
    
    subgraph "Observability"
        LS["LangSmith"]
        TR["Traces"]
        EV["Evaluations / Experiments"]
        MON["Monitoring"]
    end
    
    subgraph "Streaming"
        STR["stream() — values/updates/debug/messages"]
    end
    
    UI --> G
    G --> S
    G --> N
    G --> E
    S --> R
    N --> SEQ & PAR & COND & ITER
    G --> IMem & SQLite
    IMem & SQLite --> CP
    CP --> TT & FT
    N --> INT
    INT --> CMD & TI & SI
    N --> RC
    RC --> MA & CA
    G --> STR
    G --> LS
    LS --> TR & EV & MON

Quick Reference — LangGraph API

# Building the graph
from langgraph.graph import StateGraph, START, END

graph_builder = StateGraph(MyState)
graph_builder.add_node("node_name", node_function)
graph_builder.add_edge(START, "node_a")
graph_builder.add_edge("node_a", "node_b")
graph_builder.add_conditional_edges("node_b", routing_fn, {"label": "node_c"})
graph_builder.add_edge("node_c", END)

# Compilation and execution
graph = graph_builder.compile()                          # without persistence
graph = graph_builder.compile(checkpointer=checkpointer) # with persistence
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["node_x"],   # breakpoints before
    interrupt_after=["node_y"]     # breakpoints after
)

# Invocation
final_state = graph.invoke(initial_state)
final_state = graph.invoke(initial_state, config={"configurable": {"thread_id": "t1"}})
final_state = graph.invoke(None, config=config)  # resume / time travel

# Streaming
for chunk in graph.stream(initial_state, mode="values"):  # values / updates / debug / messages
    print(chunk)

# HITL
final_state = graph.invoke(Command(resume=human_response), config=config)

# Checkpoints
state = graph.get_state(config)
history = list(graph.get_state_history(config))

# Visualization
graph.get_graph().draw_mermaid_png()  # returns PNG bytes

Installation and Configuration

# Virtual environment
python -m venv venv
venv\Scripts\activate         # Windows
source venv/bin/activate      # Mac/Linux

# Essential dependencies (requirements.txt)
pip install langgraph
pip install langchain
pip install langchain-openai
pip install langchain-tavily
pip install langsmith
pip install python-dotenv
pip install pydantic
pip install streamlit          # for the web interface
pip install langgraph-checkpoint-sqlite  # for SQLiteSaver

# .env file
OPENAI_API_KEY=sk-...
TAVILY_API_KEY=tvly-...
LANGSMITH_API_KEY=lsv2_...
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-project

Search Terms

langgraph · essentials · developers · ai · agents · orchestration · artificial · intelligence · generative · execution · workflow · streaming · react · reducer · reducers · time · tools · travel · agent · approval · chatbot · components · core · debugging