LangGraph

Nodes, edges, state

LangChain gives you models and tools. LangGraph decides the order and keeps state. One model call with no branch does not need a graph. Reach for one when the path depends on a decision, a retry, or a person. Returning a partial dict merges. Replacing the whole message list by accident wipes history.

Starter

state = {"url": "https://example.com", "title": ""}

def write_title(state):
    return {"title": "Notes"}  # merge this key; do not rebuild the whole dict

print({**state, **write_title(state)})

A node is a function. State is one dictionary (or a typed dict) those functions read and update. An edge says which node runs after which. Because an edge can point backward, a step can repeat until the job is done.

LangChain gives you models, messages, and tools. LangGraph decides the order and keeps the state. For one model call with no branch, a plain function is simpler. Reach for a graph when the path depends on a decision, a retry, or a person.

Example: YouTube video to a blog post

The same stations show up in real jobs: invoice PDF to fields to a ledger row, a pull request to tests to a comment, classify then retrieve then draft. Put each station in its own function so a failure names the station.

Starter

from typing import TypedDict

class BlogState(TypedDict):
    url: str
    title: str

def fetch(state: BlogState) -> dict:
    return {"title": "How agents keep state"}

print(fetch({"url": "https://youtu.be/demo", "title": ""}))

Three fixed steps: pull a transcript, write a title from it, write the post from both. Nothing here needs a model to choose the next node. The graph still helps because each step has a name, a state field, and a place to retry.

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

class BlogState(TypedDict):
    url: str
    transcript: str
    title: str
    post: str

def fetch_transcript(state: BlogState) -> dict:
    # Real code calls a transcript API. This stand-in keeps the shape obvious.
    return {"transcript": f"Transcript of {state['url']}: agents, tools, and state."}

def write_title(state: BlogState) -> dict:
    return {"title": "How agents keep state"}

def write_post(state: BlogState) -> dict:
    return {"post": f"# {state['title']}\n\n{state['transcript']}"}

graph = StateGraph(BlogState)
graph.add_node("fetch", fetch_transcript)
graph.add_node("title", write_title)
graph.add_node("post", write_post)
graph.add_edge(START, "fetch")
graph.add_edge("fetch", "title")
graph.add_edge("title", "post")
graph.add_edge("post", END)
app = graph.compile()
print(app.invoke({"url": "https://youtu.be/demo", "transcript": "", "title": "", "post": ""}))

A conditional edge

Store `attempts` in state. Graphs make the loop visible. They do not make it safe. Map every return value of `route` to a real node, including the failure node.

Starter

def route(transcript: str, attempts: int) -> str:
    if attempts >= 2:
        return "give_up"
    if len(transcript) < 20:
        return "fetch"
    return "title"

print(route("", 1))

def route(state: BlogState) -> str:
    if len(state["transcript"]) < 20:
        return "fetch"   # try again, or swap in a clearer error node
    return "title"

graph.add_conditional_edges("fetch", route, {"fetch": "fetch", "title": "title"})

Cap the retries in state (attempts: int) or this edge will loop forever. Graphs make the loop visible. They do not make it safe by themselves.

A support agent that chooses tools

`create_react_agent` is a prebuilt graph: model node, tool node, loop. When refunds must not take the same path as status checks, write the nodes yourself. Streaming shows each node as it finishes, which is how you see “it refunded before it looked up the order.”

Starter

ORDERS = {"A17": "shipped on 3 March"}

def lookup_order(order_id: str) -> str:
    return ORDERS.get(order_id, "unknown")

print(lookup_order("A17"))

The user asks “where is order A17?”. One node calls the model. If the model requested a tool, a second node runs it and loops back. If the model answered in text, the graph ends.

from langgraph.prebuilt import create_react_agent
from langchain.tools import tool

ORDERS = {"A17": "shipped on 3 March", "C40": "waiting for stock"}

@tool
def lookup_order(order_id: str) -> str:
    """Look up shipping status by order id, such as A17."""
    status = ORDERS.get(order_id)
    return f"{order_id} {status}." if status else f"{order_id} is unknown."

# agent = create_react_agent(model, tools=[lookup_order])
# agent.invoke({"messages": [("user", "Where is order A17?")]})
print(lookup_order.invoke({"order_id": "A17"}))

create_react_agent is a prebuilt graph: model node, tool node, loop. When you outgrow it, write the nodes yourself so the branch is explicit — refunds go to a different node than status checks.

Pause, then resume

Compile with a checkpointer and interrupt before the dangerous node. `MemorySaver` dies with the process, which is fine for a tutorial and wrong in production. Use Postgres or Redis when more than one server handles traffic. Resume with the same thread id.

Starter

thread = {"id": "customer-asha", "status": "waiting"}
approved = True
thread["status"] = "resume" if approved else "waiting"
print(thread)

Compile the graph with a checkpointer and interrupt before the dangerous node. The process can exit. Later you resume with the same thread id and the human’s decision in state.

from langgraph.checkpoint.memory import MemorySaver

# app = graph.compile(
#     checkpointer=MemorySaver(),
#     interrupt_before=["refund"],
# )
# config = {"configurable": {"thread_id": "customer-asha"}}
# app.invoke({"messages": [...]}, config)
# later, after a person clicks Approve:
# app.invoke(None, config)   # continues from the checkpoint

Use a real checkpointer (Postgres, Redis) when more than one server handles traffic. MemorySaver dies with the process, which is fine for a tutorial and wrong for production.

Streaming (app.stream) lets a UI show each node as it finishes. That is how you debug “it refunded before it looked up the order”: you see the node order, not a single final string.

Advantages and limits

Branches, retries, and parallel steps are reviewable because they are drawn. Any chat model you can call can sit in a node. A single completion should stay a function. Prefer the message reducers in the docs when you add chat history, so you append messages instead of replacing them.

Starter

def needs_graph(steps: int, has_branch: bool) -> bool:
    return steps > 1 or has_branch

print(needs_graph(1, False))

• Branches, retries, and parallel steps are drawn, so they are reviewable.
• State, checkpoints, and human approval are built in.
• Any chat model LangChain can call can sit in a node.
• A single completion does not need a graph. Extra nodes hide simple bugs.
• You have to think in state updates. Returning a partial dict merges; replacing a whole messages list by accident wipes history. Prefer the message reducers in the docs when you add chat history.

Practice

• Add an attempts field and stop after 2 failed fetches.
• Split “refund” and “status” into two nodes with a conditional edge.
• Interrupt before refund and print the state you would show an agent in a UI.
• Name one workflow at your job that is a straight line, and one that must loop.