LangGraph core: nodes, edges, state

Three months after Arjun wrote his first agent loop, it had grown teeth. The forty tidy lines from Module 13 were now four hundred. There was an if for refunds, another for refunds above ₹500 that needed a manager, a flag called waiting_for_manager, a flag called already_refunded, and a retry path that someone (Arjun) had added at eleven one night to handle a timeout from the payments service.

On Tuesday a customer wrote in about order CR-2207, a large Diwali hamper that had arrived squashed. She got two refunds of ₹1,450 each. The payments call had timed out, the retry path re-entered the loop from the top, and nothing in the code remembered that the refund step had already run.

Meera came over with the bank statement. “Two refunds for one hamper. Why did it do that?”

Arjun scrolled through nested if statements for ten minutes. He could explain what each line did. He could not explain which path the program had taken that afternoon.

Priya pulled up a chair. “Your loop has a shape,” she said. “Read message, find order, decide, pay or escalate, reply. The shape is real. You just can’t see it, because it’s smeared across four hundred lines. Today we draw it, and then we make the drawing run.”

Why the loop got lost

Here is a trimmed piece of what Arjun’s handler had become:

def handle(message, state):
    while True:
        if state.get("needs_refund") and not state.get("waiting_for_manager"):
            if state["amount"] <= 500:
                try:
                    pay_refund(state["order_id"], state["amount"])
                except TimeoutError:
                    continue            # retry... from the very top
                state["already_refunded"] = True
            else:
                state["waiting_for_manager"] = True
        elif ...:
            ...

Look at the continue. When the payment timed out after the money had actually gone through (payment services do this), the loop jumped back to the top. already_refunded was set on the line after the call, so it was still missing. The program walked the refund path a second time.

Nothing here is exotic. The bug is that the program’s path is implicit. The order of steps lives in the arrangement of if, while and continue, and in which flags happen to be set. To answer Meera’s question (“why did it do that?”) you would have to replay the whole function in your head with the exact values from Tuesday afternoon.

A ticket moving through the kitchen

Priya drew Crumb’s kitchen on the whiteboard. “When an order comes in for a custom cake, what physically happens?”

“A ticket gets printed,” Arjun said. “It goes to the baking station. Then decorating. Then quality check. If the icing is wrong, it goes back to decorating. Otherwise packing.”

“Does the decorator need to know how the baker’s oven works?”

“No. She reads the ticket, does her bit, writes ‘decorated, rosette border’ on it, and passes it on.”

“And who decides where the ticket goes next?”

“There’s a chart on the wall. Baking always goes to decorating. Quality check is the only place with a choice.”

Priya circled three things. “The ticket is the state. Each station is a node. The arrows on the wall chart are edges. And quality check, the one place with a choice, is a conditional edge. That’s all LangGraph is.”

The analogy holds well, with one twist worth knowing now. In a real kitchen, each station holds the ticket and scribbles on it. In LangGraph, a node never edits the ticket directly. It hands back a small note (“set decision to escalated”) and LangGraph itself writes that onto the ticket. That small difference is what makes it possible to record every version of the ticket later, which you will lean on heavily in the debugging and checkpointing lessons.

The vocabulary, one word at a time

State. A Python dictionary that every node can read. You declare its shape up front with a TypedDict (a class that lists the keys and the type of each value). It is the single source of truth for one run: the customer’s message, the order ID, the decision, the reply.

Node. A plain Python function that takes the current state and returns a dictionary of updates: only the keys it wants to change. A node can be three lines of arithmetic or a call to an LLM (a large language model). LangGraph does not care which.

Edge. A fixed arrow: “after extract, always run reply”.

Conditional edge. An arrow with a decision in it. You give LangGraph a small function, called a router, that reads the state and returns a label. A mapping turns each label into the name of the next node.

START and END. Two special node names. START is where every run begins. Sending an edge to END means “this run is finished”.

StateGraph. The builder you add nodes and edges to. When you are done, you call .compile(), which checks the wiring (no edge pointing at a node that does not exist, for example) and gives you a runnable graph with .invoke() and .stream().

Show the answer

Still six keys, and only decision changed. A node’s return value is a set of updates, not a replacement state. Keys the node does not mention are left exactly as they were. This is why a node can stay small: it only talks about what it is responsible for.

Following one refund through the graph

Priya made Arjun rebuild the refund path with no LLM at all. “Pure Python first. If the mechanics are clear with boring functions, they stay clear when we swap a model in.”

Crumb’s refund policy is simple. Delivered orders up to ₹500 are refunded automatically. Anything above ₹500 goes to a manager. If the customer did not give an order number, ask for it.

Setup, if you have not already done it in Module 13:

uv init crumb-graph && cd crumb-graph
uv add langgraph langchain-anthropic python-dotenv

(With plain pip, pip install langgraph langchain-anthropic python-dotenv works too.) Put ANTHROPIC_API_KEY=... in a .env file for the model sections later; the first graph needs no key at all.

Here is the whole graph. Read it top to bottom; every piece is explained below.

# triage.py
import operator
import re
from typing import Annotated, TypedDict

from langgraph.graph import StateGraph, START, END

# Fake order database so you can run this today.
ORDERS = {
    "CR-2207": {"item": "Diwali hamper (large)", "amount": 1450},
    "CR-2214": {"item": "Eggless truffle pastry x2", "amount": 380},
}
AUTO_REFUND_LIMIT = 500  # rupees


# 1. The state: the ticket that moves between stations.
class TriageState(TypedDict, total=False):
    message: str
    order_id: str
    amount: int
    decision: str
    reply: str
    log: Annotated[list[str], operator.add]   # explained in the next section


# 2. Nodes: each takes the state and returns only what it changes.
def extract(state: TriageState) -> dict:
    match = re.search(r"CR-\d{4}", state["message"])
    if not match or match.group() not in ORDERS:
        return {"order_id": "", "amount": 0, "log": ["extract: no valid order id"]}
    order_id = match.group()
    amount = ORDERS[order_id]["amount"]
    return {"order_id": order_id, "amount": amount,
            "log": [f"extract: found {order_id} (₹{amount})"]}

def auto_refund(state: TriageState) -> dict:
    return {"decision": "refunded",
            "log": [f"auto_refund: ₹{state['amount']} refunded"]}

def escalate(state: TriageState) -> dict:
    return {"decision": "escalated",
            "log": [f"escalate: ₹{state['amount']} is above ₹{AUTO_REFUND_LIMIT}"]}

def ask_for_id(state: TriageState) -> dict:
    return {"decision": "need_id", "log": ["ask_for_id: no order number"]}

REPLIES = {
    "refunded": "Sorry about that! We've refunded ₹{amount} for {order_id}. "
                "It reaches your account in 3-5 days.",
    "escalated": "Sorry about that! A manager will review the refund for "
                 "{order_id} and reply within 24 hours.",
    "need_id": "Sorry to hear that! Could you share your order number? "
               "It looks like CR-1234.",
}

def reply(state: TriageState) -> dict:
    text = REPLIES[state["decision"]].format(
        amount=state.get("amount"), order_id=state.get("order_id"))
    return {"reply": text, "log": ["reply: sent"]}


# 3. The router: reads state, returns a label. It is not a node.
def route_refund(state: TriageState) -> str:
    if not state["order_id"]:
        return "missing"
    if state["amount"] <= AUTO_REFUND_LIMIT:
        return "small"
    return "large"


# 4. Wire it up.
builder = StateGraph(TriageState)
builder.add_node("extract", extract)
builder.add_node("auto_refund", auto_refund)
builder.add_node("escalate", escalate)
builder.add_node("ask_for_id", ask_for_id)
builder.add_node("reply", reply)

builder.add_edge(START, "extract")
builder.add_conditional_edges(
    "extract",                      # after this node...
    route_refund,                   # ...call this router...
    {"small": "auto_refund",        # ...and map its label to a node.
     "large": "escalate",
     "missing": "ask_for_id"},
)
for node in ["auto_refund", "escalate", "ask_for_id"]:
    builder.add_edge(node, "reply")
builder.add_edge("reply", END)

graph = builder.compile()

if __name__ == "__main__":
    result = graph.invoke({
        "message": "My order CR-2207 arrived squashed. The Diwali hamper was a "
                   "gift! I want my money back.",
        "log": [],
    })
    print(result["reply"])
    for line in result["log"]:
        print("  ", line)

Run it with uv run triage.py:

Sorry about that! A manager will review the refund for CR-2207 and reply within 24 hours.
   extract: found CR-2207 (₹1450)
   escalate: ₹1450 is above ₹500
   reply: sent

Here is the graph you just built:

flowchart LR
    S([START]) --> E[extract]
    E -->|small| A[auto_refund]
    E -->|large| X[escalate]
    E -->|missing| Q[ask_for_id]
    A --> R[reply]
    X --> R
    Q --> R
    R --> D([END])

Now follow the ticket by hand, exactly as LangGraph does.

Before anything runs, the state is what you passed in:

{"message": "My order CR-2207 arrived squashed. ...", "log": []}

START → extract. There is one plain edge out of START, so extract runs. The regex finds CR-2207, the lookup finds ₹1,450, and the node returns three keys. LangGraph merges them in:

{"message": "My order CR-2207 arrived squashed. ...",
 "order_id": "CR-2207", "amount": 1450,
 "log": ["extract: found CR-2207 (₹1450)"]}

The conditional edge. LangGraph calls route_refund(state). order_id is set and 1450 <= 500 is false, so it returns "large". The mapping turns "large" into "escalate". Notice that the router changed nothing in the state. Routers only choose.

escalate. Returns decision and one more log line:

{..., "decision": "escalated",
 "log": ["extract: found CR-2207 (₹1450)", "escalate: ₹1450 is above ₹500"]}

escalate → reply. A plain edge. reply reads decision, fills the template, and returns reply plus a log line. Then the edge to END finishes the run, and invoke hands you the final state.

Change the message to mention CR-2214 and the same graph takes a different branch: route_refund returns "small" (₹380), auto_refund runs, and the reply promises ₹380 back. Remove the order number entirely and you get "missing" and a polite question.

Compare this with the four hundred lines. The question “which path did it take?” now has a precise answer: a list of node names. And there is exactly one arrow into auto_refund. If refunds must never happen twice, there is one obvious place to guard: a check before that arrow. The graph did not fix the double refund on its own. It made the place where the fix belongs impossible to miss.

Reducers: how updates are merged

You may have noticed something odd. Every node returned log with one line in it, yet the final state had all of them. That is the job of the strange type on that key:

log: Annotated[list[str], operator.add]

Annotated is a standard Python way to attach extra information to a type. Here the extra information is a reducer: a function LangGraph uses to combine the old value of a key with a node’s update. operator.add on two lists means “join them”, so ["a"] + ["b"] becomes ["a", "b"].

Keys without a reducer use the default rule: the new value replaces the old one. That is right for decision and amount. It is wrong for anything that should accumulate.

Try it. Delete the Annotated[..., operator.add] and declare log: list[str]. Run again:

   reply: sent

One line. Each node’s log overwrote the previous one, and the last writer won. No error, no warning. This exact mistake will cause a real production bug in the debugging lesson, so remember the symptom: a list that should be growing only ever holds the latest item.

For conversations, LangGraph ships a reducer built for chat messages:

from langgraph.graph.message import add_messages

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

add_messages appends new messages, converts shorthand like ("user", "hi") into proper message objects, and gives each message an ID. If a node returns a message whose ID already exists, it replaces that message instead of appending. You will use that replace trick in the next lesson.

Show the answer

0 + 1450 = 1450, so this run looks fine and nothing crashes. But if any later node also returned amount (say, a node that applies a ₹50 voucher and returns amount: 1400), you would get 1450 + 1400 = 2850. Reducers apply to every update, not only the first. Give a key a reducer only when you genuinely want accumulation.

Watching it run: invoke and stream

graph.invoke(state) runs the whole graph and returns the final state. Good for scripts and tests. While you are learning, and later when you debug, you will want to see each step as it happens. That is graph.stream:

msg = "Order CR-2214 came with the pastries crushed."
for update in graph.stream({"message": msg, "log": []}, stream_mode="updates"):
    print(update)

{'extract': {'order_id': 'CR-2214', 'amount': 380, 'log': ['extract: found CR-2214 (₹380)']}}
{'auto_refund': {'decision': 'refunded', 'log': ['auto_refund: ₹380 refunded']}}
{'reply': {'reply': "Sorry about that! We've refunded ₹380 for CR-2214. ...", 'log': ['reply: sent']}}

Each item is one node’s name and exactly what it returned. That printout is the answer to “why did it do that?”, produced for free. With stream_mode="values" you get the full state after every step instead. The debugging lesson covers the other modes.

Swapping in a model

Real customers do not always say “refund”. They say “this is unacceptable, I paid for a gift”, or they ask about a late delivery, which is a different problem. Arjun wanted an LLM to classify the message before the refund path.

Here is the key point: adding a model is adding a node. The graph’s shape changes a little; the mechanics do not change at all.

# triage_llm.py
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser

from triage import (TriageState, extract, auto_refund, escalate, ask_for_id,
                    reply, route_refund, REPLIES)
from langgraph.graph import StateGraph, START, END

load_dotenv()  # reads ANTHROPIC_API_KEY from .env
llm = ChatAnthropic(model="claude-opus-5-5", max_tokens=2000)
to_text = llm | StrOutputParser()   # model reply -> plain string (next lesson explains the |)

LABELS = {"refund", "late", "other"}

class State(TriageState, total=False):
    intent: str

def classify(state: State) -> dict:
    label = to_text.invoke([
        ("system", "Label the customer message with exactly one word: refund, "
                   "late, or other. Reply with the word only."),
        ("human", state["message"]),
    ]).strip().lower()
    if label not in LABELS:      # never trust a model's label blindly
        label = "other"
    return {"intent": label, "log": [f"classify: {label}"]}

def handoff(state: State) -> dict:
    return {"decision": "handoff", "log": ["handoff: sent to a human"]}

REPLIES["handoff"] = "Thanks for writing in! A teammate will reply within a few hours."

builder = StateGraph(State)
for name, fn in [("classify", classify), ("extract", extract),
                 ("auto_refund", auto_refund), ("escalate", escalate),
                 ("ask_for_id", ask_for_id), ("handoff", handoff), ("reply", reply)]:
    builder.add_node(name, fn)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda s: s["intent"],
                              {"refund": "extract", "late": "handoff", "other": "handoff"})
builder.add_conditional_edges("extract", route_refund,
                              {"small": "auto_refund", "large": "escalate",
                               "missing": "ask_for_id"})
for node in ["auto_refund", "escalate", "ask_for_id", "handoff"]:
    builder.add_edge(node, "reply")
builder.add_edge("reply", END)
graph = builder.compile()

result = graph.invoke({"message": "Honestly unacceptable. CR-2214 was a birthday "
                                  "treat and the pastries were flattened.", "log": []})
print(result["log"])

['classify: refund', 'extract: found CR-2214 (₹380)', 'auto_refund: ₹380 refunded', 'reply: sent']

Three things to notice.

First, the model never decides how much to refund or whether to escalate. It makes one fuzzy judgement (what kind of message is this?) and plain Python does everything with money in it. That split, model for judgement and code for rules, is the most useful design habit in this module.

Second, the if label not in LABELS line. Models sometimes reply "Refund." or "refund request". The .strip().lower() handles the first; the fallback handles the rest. A label that is not in the mapping would otherwise crash the run at the conditional edge.

Third, the cost. The classify call sends roughly 80 tokens (a token is a chunk of text, about three quarters of an English word) and gets back one or two. At $4permillioninputtokensand$20 per million output tokens, that is about $0.0004, or roughly 3 paise at ₹88 to the dollar. It adds around a second of latency (waiting time) per message. Everything else in the graph is free and instant. Prices change, so check the provider’s pricing page before you budget.

The loop you wrote in Module 13 is a graph

Now the part that made Arjun sit up. Remember the hand-written agent: call the model, if stop_reason is "tool_use" run the tools and append the results, otherwise stop, with a step limit. Draw it:

flowchart LR
    S([START]) --> A[agent]
    A -->|"asked for tools"| T[tools]
    T --> A
    A -->|"finished"| D([END])

That is a graph with two nodes and one conditional edge. Here it is in LangGraph, with the tool-running node written by hand so you can see it is the same code you already know:

# agent_graph.py
import json
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

load_dotenv()
ORDERS = {"CR-1042": {"item": "Black Forest cake 1kg", "store": "Baner",
                      "status": "delayed", "eta": "tomorrow 11:00"}}

@tool
def get_order(order_id: str) -> str:
    """Look up a Crumb order by its ID, like CR-1042. Returns item, store, status and ETA."""
    return json.dumps(ORDERS.get(order_id, "not found"))

TOOLS = {"get_order": get_order}
llm = ChatAnthropic(model="claude-opus-5-5", max_tokens=2000).bind_tools(list(TOOLS.values()))

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # the ticket rail from Module 13

def agent(state: AgentState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

def run_tools(state: AgentState) -> dict:
    results = []
    for call in state["messages"][-1].tool_calls:
        output = TOOLS[call["name"]].invoke(call["args"])
        results.append(ToolMessage(content=output, tool_call_id=call["id"]))
    return {"messages": results}

def should_continue(state: AgentState) -> str:
    return "tools" if state["messages"][-1].tool_calls else "done"

builder = StateGraph(AgentState)
builder.add_node("agent", agent)
builder.add_node("tools", run_tools)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "done": END})
builder.add_edge("tools", "agent")
graph = builder.compile()

result = graph.invoke({"messages": [("user", "Where is CR-1042?")]},
                      config={"recursion_limit": 10})
print(result["messages"][-1].content)

Line for line, the pieces map across:

Module 13 hand-written loop	LangGraph
messages = [...] list	messages key with the add_messages reducer
client.messages.create(...)	the agent node
if response.stop_reason != "tool_use"	should_continue on a conditional edge
the for block in response.content tool runner	the tools node
going round the for loop again	the plain edge tools → agent
for step in range(max_steps)	recursion_limit in the config

@tool (from langchain_core.tools) turns a function into a tool the model can call, building the JSON Schema (a precise description of the inputs) from your type hints and docstring. bind_tools attaches the tools to every call the model makes. tool_calls is LangChain’s tidy list of what the model asked for, each with a name, args and id.

recursion_limit caps the number of steps (node runs) in one invocation. Hit it and LangGraph raises an error instead of looping forever: the same brakes as max_steps, now enforced by the framework.

Library APIs change quickly. If a name has moved, check the docs linked below.

Where it breaks

LangGraph is not free. Be honest about when it earns its place.

• A straight line does not need a graph. If your program is “call model, format result, send”, three function calls are clearer than a builder with three nodes and two edges. Reach for a graph when there are branches, loops, pauses or a need to inspect the path.
• The router is a single point of failure. A label missing from the mapping crashes the run at that edge. Normalise and validate every label, and always have a fallback branch.
• Type hints are not enforced. TypedDict documents the state, but Python does not check values at runtime. A node that returns amount: "1450" (a string) sails through until some later comparison fails in a confusing place.
• Reducer mistakes are silent. A missing reducer overwrites; an unwanted one accumulates. Neither raises an error.
• It is one more library to learn and keep up with. LangGraph moves quickly. Pin versions and read the changelog when you upgrade.

Try it yourself

Fix the double-refund bug properly. Add a set REFUNDED = set() and a node check_already_refunded that runs between extract and the refund router. If the order ID is in REFUNDED, route to a new already_done node whose reply says “This order has already been refunded.” Make auto_refund add the order ID to the set.

Then invoke the graph twice in a row with the CR-2214 message.

Show the expected result

The first run logs auto_refund: ₹380 refunded. The second run logs something like check: CR-2214 already refunded and replies that it has already been done. A tidy way to wire it: make check_already_refunded a node that returns {"decision": "already_refunded"} or nothing, and move the conditional edge so it starts from check_already_refunded, with the router returning "done" first if state.get("decision") == "already_refunded".

In production the “already refunded” fact must live in the order database, not a Python set, so that it survives restarts. That is exactly the kind of state you will persist in the checkpointing lesson.

Common mistakes

• Returning the whole state from a node. It works, but it hides what the node is responsible for, and with reducers it can duplicate list items. Return only the keys you change.
• Forgetting a reducer on a list. The list silently holds only the newest item. Use Annotated[list, operator.add] for logs and add_messages for chat messages.
• Putting business rules in the model. “Refund if under ₹500” belongs in Python, where it is exact and free. Ask the model only for judgements code cannot make.
• Trusting a router’s label. Strip, lowercase, check membership, fall back.
• No recursion limit thought. Any graph with a cycle needs a limit you chose on purpose.

Recap

• A LangGraph program is state (a typed dictionary), nodes (functions that return updates) and edges (what runs next).
• Conditional edges call a router that returns a label; a mapping turns labels into node names.
• Reducers decide how updates merge. The default replaces; operator.add and add_messages accumulate.
• invoke gives you the final state; stream(stream_mode="updates") shows each node’s update as it happens.
• The agent loop from Module 13 is a two-node graph: agent, tools, and a conditional edge, with recursion_limit as the brakes.

That evening, Arjun showed Meera the diagram on his screen. “This is the refund process. Each box is one step. The arrows are the only ways to move between them. Tuesday’s double refund happened because the old code could loop back to the top from the middle of a payment. In this version there is one arrow into ‘refund’, and we check before it.”

Meera studied it. “So when a customer complains, you can tell me exactly which boxes their message went through?”

“Every box, in order, with what each one decided.”

“Good. Now do that for the whole support inbox, not just refunds.” Priya, overhearing, grinned. “That’s tomorrow. Allergies, order lookups and refunds, each with its own specialist branch, and a router at the front.”

Go deeper

• LangGraph documentation: the concepts pages on state, nodes, edges and reducers.
• LangChain documentation: chat models, messages and the @tool decorator.
• Building effective agents (Anthropic): when a fixed workflow beats an open-ended agent.