LangGraph Agent Workflows

Why State Machines for Agents?

The agent loops we’ve built so far have a problem: they’re essentially while True loops with some conditional logic inside. As agents get more complex: with multiple paths, error recovery, human approvals, and long-running pauses, this becomes unmaintainable.

LangGraph solves this by modelling agent workflows as state machines: graphs where each node is a processing step and each edge is a possible transition. Instead of imperative if/else code buried in a loop, you get a visual, inspectable graph that’s easier to reason about, test, and extend.

Think of LangGraph as the same idea as a flowchart, but executable. You define the boxes (nodes) and the arrows between them (edges), and LangGraph handles the execution.

Core Concepts

State: A TypedDict that holds all information the workflow needs. Every node reads from and writes to this state.

Node: A Python function that receives the state and returns a partial state update.

Edge: A connection between nodes. Can be unconditional (always go to this node next) or conditional (go to one of several nodes based on state).

Checkpointer: Optional persistence layer that saves state at each step: allowing workflows to be paused, resumed, and inspected.

Installing LangGraph

pip install langgraph langchain-openai langchain-community

Building Your First LangGraph Workflow

Let’s build a research-and-write workflow with 3 nodes: search, evaluate, write.

First, define the state:

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

class ResearchState(TypedDict):
    goal:          str
    search_results: list[str]
    is_sufficient: bool
    draft:         str
    search_count:  int

TypedDict makes the state fully typed, you’ll get autocomplete and type checking throughout.

Now define the nodes:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model='gpt-5.6-sol', temperature=0.1)

def search_node(state: ResearchState) -> dict:
    """Search the web for information."""
    query = state['goal']
    
    # In a real implementation, call your search tool
    results = search_web(query, num_results=5)
    result_texts = [f"{r['title']}: {r['snippet']}" for r in results]
    
    return {
        'search_results': state.get('search_results', []) + result_texts,
        'search_count':   state.get('search_count', 0) + 1,
    }

def evaluate_node(state: ResearchState) -> dict:
    """Decide if we have enough information to write."""
    
    results_text = '\n'.join(state['search_results'][-10:])  # last 10 results
    
    response = llm.invoke([HumanMessage(content=f"""
Goal: {state['goal']}

Research collected so far:
{results_text}

Do we have enough information to write a comprehensive answer?
Answer with exactly: SUFFICIENT or NEED_MORE
""")])
    
    is_sufficient = 'SUFFICIENT' in response.content.upper()
    return {'is_sufficient': is_sufficient}

def write_node(state: ResearchState) -> dict:
    """Write the final answer."""
    
    results_text = '\n'.join(state['search_results'])
    
    response = llm.invoke([HumanMessage(content=f"""
Write a detailed, accurate answer to: {state['goal']}

Use this research:
{results_text}

Write 400-600 words with clear structure.
""")])
    
    return {'draft': response.content}

Connecting Nodes with Conditional Edges

# Create the graph
graph = StateGraph(ResearchState)

# Add nodes
graph.add_node('search',   search_node)
graph.add_node('evaluate', evaluate_node)
graph.add_node('write',    write_node)

# Set the entry point
graph.set_entry_point('search')

# search always goes to evaluate
graph.add_edge('search', 'evaluate')

Now add the conditional edge: the “decision diamond”:

def should_continue(state: ResearchState) -> str:
    """Route based on whether we have enough information."""
    if state['is_sufficient']:
        return 'write'   # go to writing
    elif state.get('search_count', 0) >= 3:
        return 'write'   # max searches reached, write with what we have
    else:
        return 'search'  # do another search

graph.add_conditional_edges('evaluate', should_continue, {
    'write':  'write',
    'search': 'search',
})

# write is the terminal node
graph.add_edge('write', END)

# Compile the graph
app = graph.compile()

Run the workflow:

result = app.invoke({
    'goal':          'What are the main use cases for LangGraph in production?',
    'search_results': [],
    'is_sufficient':  False,
    'draft':          '',
    'search_count':   0,
})

print(result['draft'])

Human-in-the-Loop: Approval Before Action

For high-stakes actions (sending emails, modifying databases, posting to social media), you want a human to review before the agent proceeds. LangGraph’s interrupt_before parameter pauses the graph at a specified node:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

class EmailState(TypedDict):
    task:         str
    draft_email:  str
    approved:     bool
    send_result:  str

def draft_email_node(state: EmailState) -> dict:
    """Draft the email content."""
    response = llm.invoke([HumanMessage(content=f"Draft an email for: {state['task']}")])
    return {'draft_email': response.content}

def send_email_node(state: EmailState) -> dict:
    """Send the approved email."""
    print(f"Sending email:\n{state['draft_email']}")
    return {'send_result': 'Email sent successfully'}

email_graph = StateGraph(EmailState)
email_graph.add_node('draft',  draft_email_node)
email_graph.add_node('send',   send_email_node)
email_graph.set_entry_point('draft')
email_graph.add_edge('draft', 'send')
email_graph.add_edge('send', END)

# Pause before 'send' and wait for human approval
app = email_graph.compile(
    checkpointer=checkpointer,
    interrupt_before=['send'],   # pause here
)

Run and pause:

config = {'configurable': {'thread_id': 'email-task-1'}}

# This runs to the 'draft' node, then pauses
result = app.invoke({'task': 'Follow up with the client about the proposal', 'approved': False, 'draft_email': '', 'send_result': ''}, config)

print("\n=== DRAFT EMAIL FOR REVIEW ===")
print(result['draft_email'])
print("\nApprove? (yes/no): ", end='')

Resume after human approval:

approval = input().strip().lower()

if approval == 'yes':
    # Resume from where we paused
    final = app.invoke(None, config)
    print(final['send_result'])
else:
    print("Email cancelled.")

The thread_id allows LangGraph to resume the exact right workflow state. You can pause, close your laptop, come back hours later, and resume, the state is preserved in the checkpointer.

Visualising Your Graph

LangGraph can render your graph as a PNG diagram:

from IPython.display import Image, display

# Renders a visual flowchart of nodes and edges
img = app.get_graph().draw_mermaid_png()
with open('workflow_diagram.png', 'wb') as f:
    f.write(img)
print("Saved workflow_diagram.png")

This is incredibly useful for understanding and debugging complex workflows, you can see exactly which nodes connect to which.

Exercise: Build a content moderation workflow with 3 nodes: (1) classify content as ‘safe’, ‘borderline’, or ‘harmful’, (2) for ‘borderline’ content, pause for human review with interrupt_before, (3) for ‘safe’ or human-approved ‘borderline’, publish; for ‘harmful’ or rejected ‘borderline’, reject. Test with content examples from each category and verify the human-in-the-loop step works correctly.