Module 4: Building Stateful Agents with LangGraph

From Chains to Workflows

---

Introduction: Why State Matters

In Module 3, you built agents that can call tools and interact with the world. But those agents follow a simple loop: reason, act, observe, repeat. Real-world agent workflows are more complex.

What if you need your agent to:

• Route requests to different processing paths based on classification?
• Pause and wait for human approval before taking a sensitive action?
• Loop through a refinement cycle until quality criteria are met?
• Maintain state across multiple steps with conditional branching?

This is where LangGraph shines. It lets you build agents as state machines — directed graphs where each node is a function, edges control the flow, and state persists across the entire workflow.

By the end of this module, you will:

• Understand state machines for agent workflows
• Build graphs with nodes, edges, and conditional routing
• Implement human-in-the-loop patterns
• Create a research agent with approval gates

---

4.1 State Machines for Agent Workflows

What Is a State Machine?

A state machine is a system that:

1. Holds a state (a data structure tracking what's happening)
2. Has nodes (functions that read and update the state)
3. Has edges (rules for which node runs next)
4. Has conditions (logic that determines which edge to follow)

LangGraph's Core Concepts

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

# 1. Define State — the data that flows through the graph
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    current_step: str
    result: str

# 2. Define Nodes — functions that process and update state
def step_one(state: AgentState) -> dict:
    return {"current_step": "step_one", "result": "Processed step one"}

def step_two(state: AgentState) -> dict:
    return {"current_step": "step_two", "result": "Processed step two"}

# 3. Build the Graph — connect nodes with edges
graph = StateGraph(AgentState)
graph.add_node("step_one", step_one)
graph.add_node("step_two", step_two)

graph.add_edge(START, "step_one")
graph.add_edge("step_one", "step_two")
graph.add_edge("step_two", END)

# 4. Compile and Run
app = graph.compile()
result = app.invoke({
    "messages": [],
    "current_step": "",
    "result": ""
})
print(result["result"])  # "Processed step two"

The add_messages Annotation

The Annotated[list[BaseMessage], add_messages] annotation tells LangGraph to append new messages to the list rather than replacing it. This is how conversation history accumulates:

from langchain_core.messages import HumanMessage, AIMessage

# Each node that returns {"messages": [new_message]}
# appends to the existing messages list
def chat_node(state: AgentState) -> dict:
    return {"messages": [AIMessage(content="Hello! How can I help?")]}

---

4.2 Creating Nodes and Edges

Node Anatomy

Every node is a function that:

1. Receives the current state
2. Does some work (LLM call, tool execution, data processing)
3. Returns a dictionary with updates to the state

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

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

def analyze_node(state: AgentState) -> dict:
    """Analyze the user's request and determine intent."""
    messages = state["messages"]
    last_message = messages[-1].content

    response = llm.invoke([
        SystemMessage(content="Analyze the user's request. Respond with the intent."),
        HumanMessage(content=last_message)
    ])

    return {
        "current_step": "analyze",
        "result": response.content
    }

Edge Types

LangGraph supports three types of edges:

1. Normal Edges — always go to the next node:

graph.add_edge("node_a", "node_b")  # A always flows to B

2. Conditional Edges — choose the next node based on state:

def route_function(state: AgentState) -> str:
    if state["current_step"] == "needs_review":
        return "review_node"
    return "output_node"

graph.add_conditional_edges(
    "analyze",
    route_function,
    {
        "review_node": "review",
        "output_node": "output",
    }
)

3. Entry Edge — where the graph starts:

graph.add_edge(START, "first_node")

---

4.3 Conditional Routing and Branching

Building a Content Classifier Agent

Let's build a practical example: an agent that classifies incoming requests and routes them to specialized handlers.

import os
from typing import Literal, TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

load_dotenv()

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


class RouterState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    category: str
    response: str


# --- Nodes ---

def classify_node(state: RouterState) -> dict:
    """Classify the incoming message."""
    last_message = state["messages"][-1].content

    response = fast_llm.invoke([
        SystemMessage(content=(
            "Classify this message into ONE category. "
            "Respond with ONLY the category name.\n"
            "Categories: question, complaint, feature_request, praise"
        )),
        HumanMessage(content=last_message)
    ])

    category = response.content.strip().lower()
    valid_categories = {"question", "complaint", "feature_request", "praise"}
    if category not in valid_categories:
        category = "question"

    print(f"[Classifier] Category: {category}")
    return {"category": category}


def handle_question(state: RouterState) -> dict:
    """Handle a question with a helpful answer."""
    response = llm.invoke([
        SystemMessage(content="You are a helpful assistant. Answer the question clearly and concisely."),
        *state["messages"]
    ])
    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)]
    }


def handle_complaint(state: RouterState) -> dict:
    """Handle a complaint with empathy and a solution."""
    response = llm.invoke([
        SystemMessage(content=(
            "You are a customer service agent. The user has a complaint. "
            "Respond with empathy, acknowledge the issue, and offer a solution or next steps."
        )),
        *state["messages"]
    ])
    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)]
    }


def handle_feature_request(state: RouterState) -> dict:
    """Handle a feature request by acknowledging and logging it."""
    response = llm.invoke([
        SystemMessage(content=(
            "You are a product manager assistant. The user is requesting a feature. "
            "Thank them for the suggestion, ask clarifying questions if needed, "
            "and let them know the feedback has been logged."
        )),
        *state["messages"]
    ])
    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)]
    }


def handle_praise(state: RouterState) -> dict:
    """Handle positive feedback graciously."""
    response = llm.invoke([
        SystemMessage(content=(
            "The user is giving positive feedback. Thank them warmly "
            "and let them know their feedback motivates the team."
        )),
        *state["messages"]
    ])
    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)]
    }


# --- Routing ---

def route_by_category(state: RouterState) -> Literal[
    "handle_question", "handle_complaint",
    "handle_feature_request", "handle_praise"
]:
    """Route to the appropriate handler based on classification."""
    category = state.get("category", "question")
    return f"handle_{category}"


# --- Build Graph ---

workflow = StateGraph(RouterState)

workflow.add_node("classify", classify_node)
workflow.add_node("handle_question", handle_question)
workflow.add_node("handle_complaint", handle_complaint)
workflow.add_node("handle_feature_request", handle_feature_request)
workflow.add_node("handle_praise", handle_praise)

workflow.add_edge(START, "classify")

workflow.add_conditional_edges(
    "classify",
    route_by_category,
    {
        "handle_question": "handle_question",
        "handle_complaint": "handle_complaint",
        "handle_feature_request": "handle_feature_request",
        "handle_praise": "handle_praise",
    }
)

workflow.add_edge("handle_question", END)
workflow.add_edge("handle_complaint", END)
workflow.add_edge("handle_feature_request", END)
workflow.add_edge("handle_praise", END)

router_agent = workflow.compile()


# --- Test It ---

def process_message(message: str) -> str:
    result = router_agent.invoke({
        "messages": [HumanMessage(content=message)],
        "category": "",
        "response": "",
    })
    return result["response"]


if __name__ == "__main__":
    test_messages = [
        "How do I reset my password?",
        "Your app crashed three times today and I lost my work!",
        "It would be great if you added dark mode.",
        "I love this product! Best tool I've used.",
    ]

    for msg in test_messages:
        print(f"\nUser: {msg}")
        response = process_message(msg)
        print(f"Agent: {response}")
        print("-" * 60)

---

4.4 Human-in-the-Loop Patterns

Why Humans in the Loop?

Some actions are too sensitive for an AI to execute autonomously:

• Sending emails to customers
• Deleting data
• Making purchases
• Modifying production systems

LangGraph provides built-in support for pausing a workflow and waiting for human approval using interrupts and checkpointers.

Implementing Human Approval

from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

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


class ApprovalState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    draft_email: str
    approved: bool
    sent: bool


def draft_email_node(state: ApprovalState) -> dict:
    """Draft an email based on the user's request."""
    response = llm.invoke([
        SystemMessage(content=(
            "Draft a professional email based on the user's request. "
            "Return ONLY the email content (subject and body)."
        )),
        *state["messages"]
    ])

    draft = response.content
    print(f"\n--- DRAFT EMAIL ---\n{draft}\n-------------------\n")
    return {"draft_email": draft}


def review_node(state: ApprovalState) -> dict:
    """Present the draft for human review."""
    # This node uses an interrupt to pause execution
    # In a real app, this would present a UI for approval
    print("\nThis email requires your approval before sending.")
    print(f"\nDraft:\n{state['draft_email']}\n")

    approval = input("Approve this email? (yes/no): ").strip().lower()
    approved = approval == "yes"

    return {"approved": approved}


def send_email_node(state: ApprovalState) -> dict:
    """Send the approved email."""
    print("[Email] Sending email...")
    # In production, this would call an email API
    return {
        "sent": True,
        "messages": [AIMessage(content="Email sent successfully.")]
    }


def reject_node(state: ApprovalState) -> dict:
    """Handle rejected email."""
    return {
        "sent": False,
        "messages": [AIMessage(content="Email was not sent. Let me know if you'd like to revise it.")]
    }


def route_after_review(state: ApprovalState) -> Literal["send", "reject"]:
    """Route based on approval status."""
    if state.get("approved", False):
        return "send"
    return "reject"


# Build the graph
workflow = StateGraph(ApprovalState)

workflow.add_node("draft", draft_email_node)
workflow.add_node("review", review_node)
workflow.add_node("send", send_email_node)
workflow.add_node("reject", reject_node)

workflow.add_edge(START, "draft")
workflow.add_edge("draft", "review")

workflow.add_conditional_edges(
    "review",
    route_after_review,
    {"send": "send", "reject": "reject"}
)

workflow.add_edge("send", END)
workflow.add_edge("reject", END)

# Compile with a checkpointer for state persistence
checkpointer = MemorySaver()
email_agent = workflow.compile(checkpointer=checkpointer)

Using interrupt_before for Cleaner Interrupts

LangGraph also supports declarative interrupts that pause execution before a specific node:

# Compile with interrupt_before — pauses before the "send" node
email_agent = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["send"]
)

# First invocation — runs until it hits the interrupt
config = {"configurable": {"thread_id": "email-1"}}
result = email_agent.invoke(
    {
        "messages": [HumanMessage(content="Send a follow-up email to the client about the project deadline")],
        "draft_email": "",
        "approved": False,
        "sent": False,
    },
    config=config,
)

# The agent paused before "send" — review the draft
print(f"Draft: {result['draft_email']}")

# To approve and continue:
email_agent.invoke(None, config=config)  # Resumes from checkpoint

# To reject: update state and re-invoke with a different path

When to Use Human-in-the-Loop

Scenario	Approach
Sending external communications	Always require approval
Modifying data	Require approval for destructive operations
Financial transactions	Always require approval
Internal analysis	Usually safe to run autonomously
Read-only operations	Safe to run autonomously

---

4.5 Interactive Exercise: Research Agent with Approval Gates

The Goal

Build a research agent that:

1. Takes a research topic
2. Searches the web for information
3. Drafts a summary report
4. Pauses for human approval before finalizing
5. Saves the approved report to a file

The Complete Agent

Create research_agent.py:

import os
from typing import TypedDict, Annotated, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

load_dotenv()

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


class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    topic: str
    search_results: str
    draft_report: str
    approved: bool
    final_report: str


# --- Tool ---

@tool
def web_search(query: str) -> str:
    """Search the web for information on a topic.

    Args:
        query: The search query.
    """
    try:
        from tavily import TavilyClient
        client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
        response = client.search(query, max_results=5)

        results = []
        for item in response.get("results", []):
            results.append(
                f"Title: {item['title']}\n"
                f"URL: {item['url']}\n"
                f"Content: {item['content'][:500]}\n"
            )
        return "\n---\n".join(results) if results else "No results found."
    except Exception as e:
        return f"Search error: {str(e)}"


# --- Nodes ---

def research_node(state: ResearchState) -> dict:
    """Search for information on the topic."""
    topic = state["topic"]
    print(f"[Research] Searching for: {topic}")

    results = web_search.invoke({"query": topic})
    print(f"[Research] Found results ({len(results)} chars)")
    return {"search_results": results}


def draft_node(state: ResearchState) -> dict:
    """Draft a research report from the search results."""
    topic = state["topic"]
    search_results = state["search_results"]

    print("[Draft] Writing report...")
    response = llm.invoke([
        SystemMessage(content=(
            "You are a research analyst. Write a concise research report based on "
            "the provided search results. Include:\n"
            "1. Executive Summary (2-3 sentences)\n"
            "2. Key Findings (bullet points)\n"
            "3. Analysis (1-2 paragraphs)\n"
            "4. Sources (list URLs from the search results)\n\n"
            f"Topic: {topic}\n\n"
            f"Search Results:\n{search_results}"
        )),
        HumanMessage(content=f"Write a research report on: {topic}")
    ])

    draft = response.content
    print(f"[Draft] Report ready ({len(draft)} chars)")
    return {"draft_report": draft}


def review_node(state: ResearchState) -> dict:
    """Present the draft for human review."""
    print("\n" + "=" * 60)
    print("DRAFT RESEARCH REPORT")
    print("=" * 60)
    print(state["draft_report"])
    print("=" * 60)

    approval = input("\nApprove this report? (yes/no/revise): ").strip().lower()

    if approval == "yes":
        return {"approved": True}
    elif approval == "revise":
        feedback = input("What should be changed? ")
        return {
            "approved": False,
            "messages": [HumanMessage(content=f"Please revise the report: {feedback}")]
        }
    else:
        return {"approved": False}


def finalize_node(state: ResearchState) -> dict:
    """Save the approved report."""
    report = state["draft_report"]
    topic_slug = state["topic"].lower().replace(" ", "-")[:50]
    filename = f"report-{topic_slug}.md"

    os.makedirs("reports", exist_ok=True)
    filepath = os.path.join("reports", filename)
    with open(filepath, "w") as f:
        f.write(report)

    print(f"\n[Finalize] Report saved to {filepath}")
    return {
        "final_report": filepath,
        "messages": [AIMessage(content=f"Report approved and saved to {filepath}")]
    }


def revise_node(state: ResearchState) -> dict:
    """Revise the report based on feedback."""
    feedback = state["messages"][-1].content if state["messages"] else "Improve the report."

    print("[Revise] Updating report based on feedback...")
    response = llm.invoke([
        SystemMessage(content=(
            "Revise this research report based on the feedback provided.\n\n"
            f"Current Report:\n{state['draft_report']}\n\n"
            f"Feedback: {feedback}"
        )),
        HumanMessage(content="Please revise the report.")
    ])

    return {"draft_report": response.content}


# --- Routing ---

def route_after_review(state: ResearchState) -> Literal["finalize", "revise"]:
    if state.get("approved", False):
        return "finalize"
    return "revise"


# --- Build Graph ---

workflow = StateGraph(ResearchState)

workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)
workflow.add_node("revise", revise_node)

workflow.add_edge(START, "research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", "review")

workflow.add_conditional_edges(
    "review",
    route_after_review,
    {"finalize": "finalize", "revise": "revise"}
)

workflow.add_edge("revise", "review")  # Loop back for re-review
workflow.add_edge("finalize", END)

research_agent = workflow.compile()


# --- Main ---

def main():
    topic = input("Research topic: ")

    result = research_agent.invoke({
        "messages": [HumanMessage(content=f"Research: {topic}")],
        "topic": topic,
        "search_results": "",
        "draft_report": "",
        "approved": False,
        "final_report": "",
    })

    if result.get("final_report"):
        print(f"\nDone! Report saved to: {result['final_report']}")
    else:
        print("\nReport was not approved.")


if __name__ == "__main__":
    main()

Run It

python research_agent.py

What You Built

This research agent demonstrates key LangGraph patterns:

1. Sequential Flow: Research -> Draft -> Review follows a clear sequence
2. Conditional Routing: After review, the agent either finalizes or loops back for revision
3. Human-in-the-Loop: The review node pauses execution for human approval
4. Revision Loop: Rejected reports go through a revise -> review cycle until approved
5. State Persistence: Search results, drafts, and approval status are tracked in state

This is the exact pattern used in production content generation systems, legal document review tools, and compliance workflows.

---

Key Takeaways

1. LangGraph models agent workflows as state machines with nodes, edges, and conditions
2. State is a TypedDict that flows through the graph, accumulating data at each step
3. Conditional edges let you build branching logic — route to different nodes based on state
4. Human-in-the-loop patterns are essential for sensitive actions that need human oversight
5. Revision loops let agents iterate on their output until quality criteria are met
6. Checkpointers (MemorySaver) persist state, enabling pause/resume workflows

---

Exercises

Before moving to Module 5, try these challenges:

1. Add a quality check node: Before the review step, add a node that uses an LLM to score the report quality (1-10). If the score is below 7, automatically send it back for revision without human review.

2. Multi-source research: Modify the research node to search multiple queries (the original topic + 2 related queries) and merge the results before drafting.

3. Add a citation checker: Create a node that verifies all URLs in the report are valid (actually return HTTP 200). Flag broken links before presenting to the human reviewer.

4. Parallel research: Use LangGraph's fan-out pattern to run multiple research queries in parallel and merge the results.

5. Save conversation: Add a node that saves the full conversation history (all messages) alongside the report for audit purposes.

---

Next up: Module 5, where we add memory and RAG to our agents — giving them the ability to remember conversations and reason over documents.