LangGraph in Python: State, Parallel Nodes, Checkpoints and Human Review

Anyone who has built more than a toy on top of an LLM knows three recurring pains. Calls made one after another stack up latency. A long workflow can crash and throw away all its progress. And the model will, at the worst possible moment, hand back a confident answer that's completely wrong.

LangGraph addresses all three. It's a graph-based orchestration framework built from a few plain pieces: state is a TypedDict, nodes are ordinary functions, and edges are routing logic. Those primitives combine into workflows that run nodes in parallel, checkpoint state so a crash can be recovered from, and pause for a person to review when the stakes are high.

The sections ahead go step by step. First comes one lone node, then parallel execution, conditional routing and persistent memory, followed by human-in-the-loop interrupts. A full email triage workflow at the end combines all of it. Each example is intended to run unchanged. The TypeScript SDK exposes nearly the same APIs, so the same mental model applies there.

Setup

Two packages are needed first:

pip install langgraph langchain-openai

Then environment variables for OpenAI and, optionally, LangSmith tracing:

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

The examples were tested with langgraph 1.1.3 (March 2026) and langchain-openai 1.1.10, and need Python 3.10 or later. Older guides sometimes use LANGCHAIN_TRACING_V2 and LANGCHAIN_API_KEY; both keep working, but LANGSMITH_TRACING and LANGSMITH_API_KEY are now the preferred names.

Everything in LangGraph begins with two ideas: state holds the data, and nodes are functions that transform it. Once those are comfortable, the rest of the framework falls into place.

Defining state with TypedDict

State in LangGraph is a plain Python TypedDict, with no special base classes and no framework-specific decorators, just typed dictionaries:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

The class declares the shape of the data flowing through the graph, and every node reads from and writes to this shared structure.

Why a TypedDict and not a regular dict or a dataclass? It gives static type checking, since mypy and Pyright both understand it, along with type hints, and it avoids the runtime overhead of a dataclass or a Pydantic model. LangGraph can also inspect the annotations to learn the schema, which matters for structured output and validation.

The line to understand is Annotated[list[str], operator.add], which declares a reducer. It's the most important concept in LangGraph state, and it trips almost everyone up the first time.

Without a reducer, a node returning {"summary": "new value"} simply replaces the stored value. For scalar fields like summary that is what you want. Trouble starts when two parallel nodes both write to messages: whichever finishes last is the only one kept. Declaring operator.add instead tells LangGraph to append, so every contribution to messages accumulates safely.

More complex merge logic calls for a custom reducer:

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

This reducer joins the existing list with the new messages and then trims it to the ten most recent, which helps with long conversations where token counts would otherwise balloon. operator.add covers simple appending. A custom reducer is the choice for deduplicating, capping list size or any merge beyond plain appending.

The classic mistake is forgetting a reducer on a field that parallel nodes write to. Without one, the branches silently overwrite each other, the last to finish wins, and no error ever appears.

TypedDict versus Pydantic

Both show up in LangGraph, and the split between them is deliberate. TypedDict is for graph state, the container that moves through the graph. LangGraph was designed around it, its fields support the Annotated reducer syntax, it serializes cleanly for checkpointing and it adds no runtime overhead.

Pydantic is for schemas that face the LLM. Calling .with_structured_output(SomeModel) gives the LLM a JSON schema to aim at and validates that the reply fits. A response that doesn't match raises an error immediately and doesn't let malformed data travel through the graph.

The capstone workflow later follows exactly this division: EmailState is a TypedDict serving as the graph container, and EmailClassification is a Pydantic model describing the LLM's output.

Writing a first node

Each node takes the full state as input and hands back a partial update dict. LangGraph folds that update into the state and applies any reducers that were declared:

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

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

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Going through the code:

• StateGraph(ChatState) creates a graph that uses ChatState as its schema, and LangGraph reads the annotations to learn about reducers and field types.
• graph.add_node("chatbot", chatbot) registers the function under the name "chatbot", which is what edge definitions refer to.
• graph.add_edge(START, "chatbot") makes "chatbot" the first node to run. START is a built-in sentinel.
• graph.add_edge("chatbot", END) ends execution after chatbot finishes. END is another built-in sentinel.
• graph.compile() validates the structure, checking for dangling edges and unreachable nodes, and returns a runnable CompiledGraph.
• app.invoke({"messages": [...]}) runs the graph synchronously and returns the final state.

The node wraps its string in a list since messages uses operator.add as the reducer. Returning {"messages": [response.content]} tacks a single-item list onto the existing one, while returning the bare string would fail at runtime because operator.add cannot add a string to a list.

The rule is to match what a node returns to the type of the field, reducer included. A list[str] field with operator.add takes a list[str], and a plain str field with no reducer takes a str.

That's a complete working LangGraph application in about 20 lines. What happens inside invoke() is the next question.

Super steps: how execution works

LangGraph runs nodes in super steps. Every node scheduled for the same step runs concurrently, and all of them must finish before the next step starts. The single-node example has one super step holding one node. Adding parallel edges puts several nodes in the same super step.

This matters for two reasons. Parallel execution is built into the model from the start and isn't something bolted on later. And with a checkpointer, LangGraph saves a complete state snapshot after each super step finishes. That's what makes crash recovery possible: if the process dies midway, execution resumes from the last completed super step.

With state flowing through a single node understood, the next question is how to control order or run nodes at the same time. That's the job of edges.

Edges: serial, parallel and conditional

Edges control how execution moves between nodes. LangGraph offers three types, and between them they cover every routing pattern that comes up. The simplest form already appeared in the chatbot example.

Serial edges

The simplest case is one node followed by the next.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft runs first. Once it finishes and LangGraph merges the draft field into state, review_draft runs and can read the draft from state["draft"]. Serial ordering is a guarantee, because the edge enforces the dependency, and no sequencing code is needed. Serial edges fit whenever node B depends on node A's output: generate then review, fetch then process, classify then route.

Accidental cycles are a common trap. By default LangGraph expects a directed acyclic graph, so adding an edge from review back to generate raises an error at compile time unless the graph is explicitly set up to be cyclic. Iterative loops are usually built with conditional edges that can route back to an earlier node.

Parallel edges

When two edges fan out from one node, their targets run concurrently in the same super step. Reducers become essential here:

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Two edges from START schedule research and outline in the same super step, and LangGraph runs them concurrently. Both also have an edge to combine, and LangGraph won't schedule combine until every node pointing at it has finished, which acts as an automatic synchronization barrier.

findings has a reducer even though only research writes to it here, because a real system might have four parallel research agents appending to the same list, and without the reducer only one result would survive.

The latency math is simple. If research takes 2 seconds and outline takes 3, the serial version needs 5 seconds and the parallel version needs 3, the length of the longer task. For independent LLM calls, parallel execution is close to free latency savings.

Candidates for parallelization are all pairs of nodes that don't depend on each other's output. Even moving two one-second calls into the same super step adds up across thousands of requests. Watch for nodes that look independent but share state in less obvious ways. If two parallel nodes write to one field without a reducer, that's a race condition, the last to finish decides the value, and LangGraph gives no warning.

This fan-out and fan-in pattern is the backbone of the email capstone, which runs a knowledge-base search and a bug-tracker lookup in parallel before drafting a response.

Conditional edges

When the next node depends on what the current one produced, add_conditional_edges applies:

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

The function route_by_sentiment looks at the state and returns a string key, and the mapping dictionary given to add_conditional_edges turns that key into a node name. A routing function that stays pure (state in, string out, no side effects) is easy to test, because it can be unit tested with no LLM involved.

add_conditional_edges takes three arguments: the source node, the routing function, and a dictionary mapping string keys to destination node names. Every key the routing function might return must appear in that dictionary, or LangGraph raises an error at runtime.

The Command alternative

When a node needs to update state and route in the same step, it can return a Command directly and skip add_conditional_edges:

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

One Command return value carries both a state update (update=) and a routing choice (goto=), so the node writes data and picks the next destination in a single move.

add_conditional_edges is the right pick when the routing logic only reads state and writes nothing new, when routing should stay separate from node logic, or when several source nodes share the same routing function.

Command is the right pick when the node produces new data that influences its own routing, when a one-off case doesn't justify a separate routing function, or when the destination depends on something computed inside the node and not already in state.

For the human review gate the email capstone relies on add_conditional_edges, since the routing there is pure: it checks only the priority and confidence already stored in state.

Memory and checkpointers

With ordering and branching under control, the next problem is persistence: what happens when the process restarts in the middle of a conversation?

Until now every invocation has started from nothing. Conversations and multi-step workflows need state that survives between calls. LangGraph handles this with checkpointers, and the API is clean enough that moving from development to production persistence is a one-line change.

MemorySaver for development

MemorySaver keeps state in memory, which suits development and testing:

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

With checkpointer=memory, the compiled graph stores a full snapshot after every super step. A repeat invoke() call using the same thread_id makes LangGraph load the latest checkpoint for that thread first. The thread_id works like a session ID: reuse it to continue, change it to start clean.

The second call can leave out earlier messages because the checkpointer holds them already. If Alice invokes with {"messages": ["What should I learn first?"]}, LangGraph loads the "user-alice" checkpoint, finds the stored list and appends the new entry through the operator.add reducer. Once chatbot runs, state["messages"] contains the whole history.

Reusing one thread_id for unrelated conversations is a mistake. If "user-alice" is reused for a different topic, the chatbot arrives with all of Alice's earlier conversation in context, so each new session should get a fresh thread_id. During development, app.get_state_history(config) shows the full checkpoint history for a thread, and the same mechanism powers time-travel debugging in LangSmith.

Durable checkpointers for production

MemorySaver lives in process memory, so when the process dies the state goes too. Production needs a durable backend:

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() creates the required database tables on the first run and is a no-op afterward, so it only needs calling once during initialization. The API does not change between backends, since checkpointers sit behind a shared abstraction. MemorySaver, PostgresSaver and RedisSaver follow one protocol, so the graph code has no idea which of them it is talking to.

Choosing between them:

• MemorySaver fits development, testing and demos, where surviving a restart doesn't matter.
• PostgresSaver fits production workflows that need durability and the ability to query checkpoint history.
• RedisSaver, from langgraph-redis, fits high-throughput cases that need fast checkpoint reads and can accept Redis's persistence tradeoffs.

What gets saved

A full snapshot is stored at every super step. That enables time travel: the state at any point in the history can be inspected, replayed from a chosen checkpoint, or debugged to see exactly where things went wrong. The same mechanism underpins interrupts. When a node calls interrupt(), LangGraph needs somewhere to keep the paused state so it can resume later. With no checkpointer, nothing can hold it, and the interrupt pattern collapses.

Human-in-the-loop with interrupts

Some decisions are too important to automate. Before an agent sends a customer email, approves a refund or deploys code, a person should review and approve. LangGraph handles that with the interrupt() function.

The interrupt() function

Placing interrupt() inside a node halts execution, writes state to the checkpointer and hands a payload back to the caller:

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

In order, the call does four things:

1. It freezes execution at the exact line where it's called.
2. It saves the complete state to the checkpointer, which is why one is required.
3. It returns the payload that was passed in, here the draft text, to whoever called invoke().
4. It waits, leaving the graph suspended.

The decision variable receives nothing at first, because interrupt() doesn't return until the graph is resumed with Command(resume=value). At that point decision gets whatever was passed to resume=.

The payload tells the human reviewer what they're looking at, so it can carry whatever they need: the draft, a summary, or a structured dict with context. The caller gets it as part of the interrupt notification.

Resuming with Command

Continuing after an interrupt means invoking the graph again with Command(resume=value):

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

In the first step, LangGraph runs generate, stores the draft in state and reaches interrupt() in the review node. It checkpoints the state and returns, and the result from this first invoke() holds the interrupt payload, the draft for the reviewer. In the second step, Command(resume="approve") makes LangGraph load the checkpoint for "email-review-1", resume at the interrupt() call and hand back "approve" as its return value. The review node's decision becomes "approve", it sets approved=True and execution moves on to send.

The same config must be used for both calls, since LangGraph finds the right checkpoint through thread_id. A different thread_id on the second call would point at a checkpoint that doesn't exist, and the call would fail.

A checkpointer is mandatory for interrupts. Without one the paused state has nowhere to go, so app.compile(checkpointer=memory) isn't optional here.

Since LangGraph 1.1.0, invoke() accepts version="v2" and returns a GraphOutput with .value and .interrupts attributes, giving type-safer interrupt handling worth trying once the basics are familiar. An interrupt payload can be any serializable value, whether string, dict or list. Production setups tend to pass a structured dict holding everything a reviewer needs, such as draft, classification, sender and review URL, which the application layer can render in a UI or forward through Slack or email.

There is one more pitfall. Calling invoke() again with the full initial state in place of Command(resume=value) is a mistake: app.invoke({"messages": [...]}, config=config) on the same thread_id restarts the graph for that thread and discards the paused state. An interrupted graph should only ever be continued with Command(resume=value).

Capstone: an email triage workflow

With the pieces in place, they can be merged into a production-style email triage system. It reads an incoming email, classifies the intent via structured output, searches a knowledge base and a bug tracker in parallel, drafts a reply, and can stop for human review before sending.

Each earlier concept maps onto a part of this workflow:

• TypedDict state with reducers shows up as EmailState, with operator.add on search_results.
• Serial edges link read_email to classify_intent.
• Parallel edges let classify_intent branch into search_docs and bug_tracking.
• Fan-in synchronization means both parallel nodes feed write_response.
• Conditional routing is route_by_priority, which chooses between human review and auto-send.
• Pydantic structured output is EmailClassification, which constrains the LLM's reply.
• The interrupt is human_review, which pauses for approval.
• The checkpointer is MemorySaver, which enables the interrupt and resume cycle.

State and schemas

The state and the Pydantic schema for structured classification come first:

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification is a Pydantic model and not part of EmailState, which is the TypedDict and Pydantic split in action. It's the shape handed to the LLM as a JSON schema target, and Pydantic validates the response and raises immediately if it doesn't match. EmailState is the graph container, a TypedDict because LangGraph needs the reducer syntax and efficient serialization.

classification is stored as a dict in EmailState because, after Pydantic validates the response, .model_dump() converts it to a plain dict before it goes into state. That keeps state serializable by any checkpointer, with no Pydantic dependency needed at restore time.

The field that matters most is search_results: Annotated[list[str], operator.add]. Since search_docs and bug_tracking run in parallel and can both write results, the reducer makes their outputs pile up together, whereas without it one branch would quietly overwrite the other.

Building the nodes

Nodes stay small and single-purpose. Each pulls only the values it needs out of state, does its job and returns a partial update:

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

At first glance read_email seems pointless, as it only returns fields already present in state, but it acts as an explicit parsing step. A production version would take a raw webhook payload and pull out the fields, so the function here stands in for that extraction.

classify_intent is where Pydantic proves useful. In structured_llm = llm.with_structured_output(EmailClassification), the LLM is wrapped so replies must obey the schema. It receives a JSON schema derived from EmailClassification and must return matching valid JSON, and LangChain retries automatically before raising an error if it does not. .model_dump() then converts the validated object into a plain dict for state.

The two parallel nodes write to separate fields, search_results and bug_info, so their LLM calls run at once. For search_results, search_docs returns a list ([f"[Docs] {response.content}"]), as operator.add needs list inputs. bug_tracking returns a plain string for bug_info, which has no reducer, so ordinary assignment is enough.

The fan-in node is write_response. When it starts, both parallel nodes are done and their results sit in state, so it builds the prompt from everything and drafts the reply.

human_review passes a structured dict to interrupt() and not a plain string, which is good production practice. The dict holds the draft, the classification and a message for the reviewer, so whatever system sits on the other side, whether a web UI, a Slack bot or an email notification, gets the whole context.

send_reply uses state.get("approved", True) as a defensive default, so if approved is somehow missing from state it sends. That default can flip to False where the risk profile leans cautious.

In production, search_docs and bug_tracking would call real systems such as a vector database, Jira or GitHub Issues, and the LangChain @tool decorator makes wrapping those calls easy. The node structure stays the same, with a tool call replacing the LLM call. One more trap: write_response reads state["search_results"], so if that's still an empty list the access can fail, which is avoided by initializing search_results: [] in the initial invoke payload so the reducer starts appending from there.

Wiring the edges

Serial, parallel and conditional edges then connect the nodes:

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority sends high-priority emails to human review regardless of confidence. Classifications with confidence under 0.8 also go to review, because if the model isn't sure what an email is about, a person should decide. Everything else is sent automatically, which is a reasonable production default to be tuned against real error rates.

This routing goes through add_conditional_edges and not Command, because route_by_priority is a pure function: it only inspects state and returns a key, creating no new data, which is exactly the case add_conditional_edges is meant for.

Here is every connection in the graph:

• A serial run goes from START through read_email into classify_intent.
• Fan-out from classify_intent launches both search_docs and bug_tracking together.
• Fan-in sends the results of those two nodes on to write_response.
• After write_response, a conditional edge picks human_review or send_reply.
• Once review is done, human_review feeds send_reply, which ends at END.

Running the workflow end to end

A sample email goes in first, and the first invocation runs until the interrupt:

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

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

By then the graph has done a lot of work:

1. read_email parsed the message.
2. classify_intent labeled it a high-priority bug report using structured Pydantic output.
3. The docs search and the bug-tracker lookup ran in parallel within one super step.
4. write_response drafted a reply that used all of that context.
5. The human_review node called interrupt() and paused.

The whole processing pipeline, including both parallel LLM calls, finished before the pause. When the graph stops, state is fully saved, so the reviewer can take minutes or hours to respond without losing anything.

The reviewer then approves:

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph loaded the checkpoint for "email-001", resumed human_review from its interrupt() call, passed "approve" back as the return value, set approved=True in state and carried on to send_reply, which returned the confirmation status.

Every earlier idea appears in the capstone: reducer-backed TypedDict state, serial and parallel edges, conditional routing, Pydantic structured output, human review through an interrupt and memory kept by a checkpointer.

For a low-priority email with high confidence, route_by_priority returns "auto_send" and human_review is skipped. The graph runs straight through, reading, classifying, searching in parallel, drafting and sending, and the interrupt only exists on the path that needs it. That's the payoff of conditional routing.

A closing caution concerns initial state. TypedDict sets no defaults for missing fields, unlike Pydantic. If write_response reaches for state["search_results"] and the key wasn't in the initial state, a KeyError follows. Every field belongs in the initial state dict, even if it starts as an empty string, an empty list or False.

Where to go next

The route taken here begins with a single node and a TypedDict, and finishes with a production-style workflow that can run LLM calls side by side, validate structured output, branch conditionally, survive crashes through checkpoints and wait for a reviewer. A few directions lead on from there:

• LangGraph Platform handles deployment, managed checkpointing and horizontal scaling, which removes the burden of operating the infrastructure when real traffic is the goal.
• LangSmith traces and debugs every graph execution and shows the exact state at each super step when a node misbehaves. That becomes important once graphs reach five or more nodes with conditional paths.
• Subgraphs compose smaller graphs into larger multi-agent systems. Each manages its own state and talks through a parent graph, a natural step after single-graph workflows.
• Tool calling through LangChain @tool and ToolNode lets agents use external APIs, databases and services from within the graph, and the capstone nodes search_docs and bug_tracking are obvious places to try it.
• Production persistence means installing langgraph-checkpoint-postgres for durable state that survives restarts, and the API swap is a single line.

Quick recap

• State is a TypedDict. Any field typed Annotated[list, operator.add] collects values across writes, and a field with no reducer gets replaced on every write.
• A node is just a function: it gets the full state and returns a partial update dict, with LangGraph doing the merge.
• Edges decide the flow. Use serial edges where order matters, parallel edges for independent work running together, and conditional edges when the data should pick the branch.
• Use a TypedDict for the graph state, and use Pydantic with .with_structured_output() to pin down what the LLM returns.
• Checkpointers provide persistence and make interrupts possible: MemorySaver while developing, PostgresSaver in production. No checkpointer, no interrupts.
• An interrupt is nothing more than saved, paused state. interrupt(payload) halts and checkpoints, and Command(resume=value) restarts from the exact spot.
• Parallelism comes built in: two edges leaving one node run their targets at once, and every field those nodes write needs a reducer.

Questions to think through

1. MemorySaver fits development and PostgresSaver fits production. Which other storage backends deserve a look for a high-throughput system, and what tradeoffs would settle the choice?
2. route_by_priority sends high-priority emails to review and auto-sends the rest. How could that threshold be tuned in a real system, and which metrics would show whether it's set correctly?
3. The capstone routes after write_response with add_conditional_edges. Is there a scenario in the email workflow where the Command pattern would fit better than a separate routing function?