Day 20 Theory — Why a Graph Runtime? LangGraph State, Nodes and Edges

Lesson goal

In the theory section you learned why a graph runtime becomes useful. Now we will implement that idea using the real LangGraph Python package.

By the end of the lesson you should be able to:

• define typed shared state,
• attach reducers to fields that need merge behavior,
• implement nodes that return partial state updates,
• add normal and conditional edges,
• compile and invoke a real StateGraph,
• inspect streamed node updates,
• explain how termination and recursion limits protect loops,
• compare graph orchestration with our earlier hand-written loop.

What changed in the Service Desk today?

Yesterday our orchestration was primarily imperative.

Today the same major steps become explicit graph nodes:

graph TD
    START --> classify
    classify -->|needs KB| retrieve
    classify -->|no KB| decide
    retrieve --> decide
    decide -->|tool needed| tool
    decide -->|done| finalize
    tool -->|continue| decide
    tool -->|done| finalize
    finalize --> END

This graph contains a real cycle (decide -> tool -> decide) and explicit termination paths.

Step 1 — Use the real LangGraph package

The Day 20 environment now depends on a pinned current course version of LangGraph:

langgraph==1.2.11

The graph primitives come from the actual library:

from langgraph.graph import StateGraph, START, END

We are not implementing a home-grown StateGraph clone. The point of this day is to learn LangGraph itself after understanding the lower-level agent loop.

Step 2 — Define the shared state

Our graph state is a TypedDict.

A shortened example:

import operator
from typing import Annotated, TypedDict

class ServiceDeskState(TypedDict, total=False):
    user_id: str
    query: str
    messages: Annotated[list[dict], operator.add]
    node_history: Annotated[list[str], operator.add]
    iteration_count: int
    max_iterations: int
    final_response: str

Why total=False?

Not every field exists at the first node. Classification, tool results and the final answer are produced later.

Why Annotated[..., reducer]?

Without a reducer, a field normally follows replacement semantics when a node returns a new value.

For append-only history, we want:

["classify"] + ["retrieve"] -> ["classify", "retrieve"]

not:

["classify"] replaced by ["retrieve"]

The reducer defines that merge rule.

Step 3 — Write nodes as state transformations

A graph node receives current state and returns a partial update.

def classify_node(state: ServiceDeskState) -> dict:
    query = state.get("query", "")
    classification = classify(query)

    return {
        "classification": classification,
        "node_history": ["classify"],
    }

Notice what the node does not do:

state["classification"] = classification
return state

Returning a focused update makes each node's responsibility clearer and lets LangGraph apply the state schema/reducer rules.

Step 4 — Build the graph

The course implementation now uses:

from langgraph.graph import END, START, StateGraph

builder = StateGraph(ServiceDeskState)

builder.add_node("classify", classify_node)
builder.add_node("retrieve", retrieve_node)
builder.add_node("decide", decide_node)
builder.add_node("tool", tool_node)
builder.add_node("finalize", finalize_node)

builder.add_edge(START, "classify")

At this point we have registered executable node functions and the entry edge.

We have not executed anything yet.

Step 5 — Add conditional routing

Classification decides whether retrieval is needed.

def route_after_classification(state: ServiceDeskState) -> str:
    if state["classification"].requires_kb:
        return "retrieve"
    return "decide"

Then wire the route:

builder.add_conditional_edges(
    "classify",
    route_after_classification,
    {"retrieve": "retrieve", "decide": "decide"},
)

Why is this better than burying the if inside another giant node?

Because the transition is now part of the graph topology. We can inspect and test it independently.

Step 6 — Create the agent cycle

Our decision node can route to a tool or to finalization:

decide
  |\
  | \-- final answer --> finalize
  |
  \---- tool ----------> tool
                           |
                           +----> decide again

The cycle exists because after observing a tool result the agent may need another decision.

That is the graph form of the loop we built earlier.

Step 7 — Add an explicit terminal path

builder.add_edge("finalize", END)

The graph is allowed to cycle, but successful workflow paths eventually reach END.

Application routing also tracks iteration_count / max_iterations so business logic can stop repeated tool behavior deliberately.

LangGraph additionally provides recursion-step protection as a runtime safety boundary.

Step 8 — Compile the graph

graph = builder.compile()

Remember the distinction:

StateGraph builder
      |
      | add nodes/edges
      v
compile()
      |
      v
compiled runnable graph

We invoke the compiled graph, not the builder definition.

Step 9 — Invoke it

initial_state = {
    "user_id": "alice_99",
    "query": "I need to reset my corporate password",
    "messages": [
        {"role": "user", "content": "I need to reset my corporate password"}
    ],
    "node_history": [],
    "tool_calls": [],
    "tool_results": [],
    "retrieved_docs": [],
    "iteration_count": 0,
    "max_iterations": 5,
    "is_complete": False,
}

result = graph.invoke(initial_state)

The returned state contains the merged updates produced along the selected graph path.

For a password-reset example we expect a history similar to:

classify
retrieve
decide
tool
decide
finalize

Step 10 — Stream graph updates

LangGraph can stream updates while execution progresses.

The course agent uses:

for event in graph.stream(initial_state, stream_mode="updates"):
    print(event)

An update event lets us observe which node produced a state delta without inventing our own tracing protocol.

This is useful for:

• debugging,
• progress UI,
• demonstrations,
• testing node order.

It is not a reason to expose private model reasoning.

Reducer experiment

A small real-LangGraph test makes reducer behavior visible:

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

class State(TypedDict, total=False):
    count: int
    items: Annotated[list[str], operator.add]

builder = StateGraph(State)

builder.add_node("one", lambda state: {"count": 1, "items": ["A"]})
builder.add_node("two", lambda state: {"count": 2, "items": ["B"]})
builder.add_edge(START, "one")
builder.add_edge("one", "two")
builder.add_edge("two", END)

graph = builder.compile()
print(graph.invoke({"count": 0, "items": []}))

Expected idea:

count -> 2          # replacement
items -> ["A", "B"] # reducer merge

This is much easier to remember after seeing both semantics in one state object.

Recursion protection experiment

Create an intentionally invalid infinite graph:

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START
from langgraph.errors import GraphRecursionError

class LoopState(TypedDict):
    tick: Annotated[list[int], operator.add]

builder = StateGraph(LoopState)
builder.add_node("a", lambda state: {"tick": [1]})
builder.add_node("b", lambda state: {"tick": [2]})
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "a")
loop_graph = builder.compile()

try:
    loop_graph.invoke({"tick": []}, {"recursion_limit": 10})
except GraphRecursionError:
    print("Graph exceeded its allowed step budget")

The lesson is not “set recursion limit to 10.”

The lesson is:

A cyclic workflow needs meaningful termination logic, and the runtime should also have a bounded safety limit.

Debugging checklist

If the graph takes the wrong path, inspect in this order:

1. What state entered the routing function?
2. What route value did it return?
3. Is that route present in the path map?
4. What update did the previous node return?
5. Did a reducer merge the update differently than expected?
6. Is a required state field missing?
7. Is the graph looping because the business termination condition never becomes true?

Do not start by increasing recursion limits.

Common mistakes

Mistake 1 — believing every list automatically appends

Reducer behavior must be declared for fields that need merging.

Mistake 2 — mutating all graph state in place

Prefer node return updates so state ownership remains clear.

Mistake 3 — using a graph for a trivial linear function

If your workflow is simply:

validate -> call one API -> return

ordinary Python may be clearer.

Mistake 4 — assuming compile means execute

compile() creates an executable graph. invoke, ainvoke or streaming APIs run it.

Mistake 5 — creating a cycle with no business stop condition

A recursion limit protects the runtime; it does not define successful completion.

Practical lab

Work in:

service-desk-day-20/

A. Verify the dependency

python -c "import langgraph; print(langgraph.__file__)"

The implementation should use the installed LangGraph package rather than a local clone.

B. Inspect service_desk/models.py

Identify:

• replacement fields,
• reducer-backed fields,
• why node_history appends,
• why retrieved documents use a deduplicating reducer.

C. Inspect service_desk/graph.py

Draw the exact graph topology from the code before running it.

Then compare your drawing with:

graph = build_service_desk_graph()
print(graph.get_graph())

D. Run one request

from service_desk.agent import ServiceDeskGraphAgent

agent = ServiceDeskGraphAgent()
result = agent.handle_request(
    "student_01",
    "I need to reset my corporate password",
)

print(result["node_history"])
print(result["final_response"])

Explain why each node ran.

E. Stream the same request

Use stream_request() and record the node update order.

F. Break the termination condition intentionally

Use the recursion-protection test to observe GraphRecursionError.

Then explain why raising the limit is not the correct fix for a logically infinite graph.

G. Run tests

cd service-desk-day-20
PYTHONPATH=. pytest -q

The important tests now prove behavior against the real LangGraph runtime.

Knowledge check

1. What is the difference between a node and an edge?

A node performs work and returns state updates. An edge determines which node executes next.

2. Why use a reducer?

To define how multiple updates to one state field should be combined instead of relying on replacement semantics.

3. What does compile() do conceptually?

It turns the declared graph definition into an executable runnable graph.

4. Why can decide -> tool -> decide be correct?

It represents the agent decision/action/observation cycle. It is safe only when meaningful termination conditions and runtime limits exist.

5. Does LangGraph replace our earlier agent loop knowledge?

No. It represents that control flow explicitly and adds runtime facilities around it.

Scenario

A production graph repeatedly hits GraphRecursionError after the maximum number of steps.

A teammate proposes increasing recursion_limit from 25 to 500.

What should you investigate first?

Answer: Inspect the state changes and routing decisions to determine why the business termination condition is never reached. A higher limit may only make an infinite or non-progressing workflow more expensive.

Key takeaways

• We now use the actual LangGraph runtime, not a course-made imitation.
• StateGraph makes workflow state and transitions explicit.
• Nodes return partial updates; reducers define merge semantics.
• Conditional edges make branching inspectable and testable.
• Cycles are useful for agent behavior but require explicit termination.
• Compilation and execution are separate concepts.
• Use a graph when it improves control-flow clarity—not simply because the application contains an LLM.

Official references

• LangGraph Graph API: https://docs.langchain.com/oss/python/langgraph/graph-api
• LangGraph Graph API examples: https://docs.langchain.com/oss/python/langgraph/use-graph-api
• LangGraph Quickstart: https://docs.langchain.com/oss/python/langgraph/quickstart