AI & ML / Agents & orchestration / LangGraph / 03_nodes_edges_and_routing.md

Nodes, edges and routing

Updated 6 interview angles 4 min read source
On this page7
  1. The smallest complete graph
  2. Three kinds of edge
  3. Loops
  4. Parallel fan-out and fan-in
  5. Node signatures
  6. Related
  7. Interview angle

Nodes, edges and routing

Nodes do the work; edges decide what runs next. This note is the whole control flow surface.

The smallest complete graph

python 
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class State(TypedDict):
    question: str
    answer: str

def answer(state: State) -> dict:
    return {"answer": f"echo: {state['question']}"}

builder = StateGraph(State)
builder.add_node("answer", answer)
builder.add_edge(START, "answer")
builder.add_edge("answer", END)

graph = builder.compile()
graph.invoke({"question": "hi"})

You must compile before invoking. compile() validates the graph — every node reachable, every edge target real — and returns the runnable. It is also where a checkpointer is attached, which is why persistence is a compile-time decision rather than a per-call one.

START and END are sentinels, not nodes. START marks the entry; reaching END finishes the run.

Three kinds of edge

Edge When the target is known
Normal at build time
Conditional at run time, by a function
Command at run time, by the node itself

Normal

python 
builder.add_edge("retrieve", "generate")

Always run generate after retrieve. Two normal edges out of one node means both targets run in parallel — that is how you fan out, and it is why parallel writes need reducers.

Conditional

A router function reads state and returns the name of the next node:

python 
from typing import Literal

def route(state: State) -> Literal["retry", "finish"]:
    if state["score"] < 0.5:
        return "retry"
    return "finish"

builder.add_conditional_edges("check", route)

The router is plain Python — it does not call a model. That is the point: put the branch in code where it is deterministic and testable, and reserve the model for the decisions that genuinely need judgement.

Annotate the return with Literal so the possible targets are visible to the graph and to whoever reads it.

Command — update and route together

python 
from typing import Literal
from langgraph.types import Command

Lanes = Literal["urgent", "normal"]

def triage(state: State) -> Command[Lanes]:
    lane = classify(state["ticket"])
    return Command(
        update={"lane": lane},
        goto="urgent" if lane == "p1" else "normal",
    )

One node, one return, both a state update and the next hop. Use it when the routing decision is a by-product of the work the node just did — splitting that into a node plus a router means recomputing or stashing the answer.

Loops

A cycle is just an edge pointing backwards:

python 
builder.add_edge("generate", "check")
builder.add_conditional_edges("check", route)
# route returns "generate" to loop, or "finish"

That is the agent loop. Nothing special about it — which is the insight worth carrying: an agent is a graph with a cycle and a model in the router.

Gotcha: every cycle needs a termination condition you control. The runtime enforces a recursion limit as a backstop, and hitting it is an error, not a graceful stop. Put an explicit step counter in state and route to END when it is exceeded, so exhaustion is a decision rather than a crash. See Agent failure modes.

python 
Next = Literal["generate", "finish"]

def route(state: State) -> Next:
    # Your limit, not the runtime's.
    if state["steps"] >= 6:
        return "finish"
    if state["score"] < 0.5:
        return "generate"
    return "finish"

Parallel fan-out and fan-in

python 
builder.add_edge("plan", "search_web")
builder.add_edge("plan", "search_docs")
builder.add_edge("search_web", "merge")
builder.add_edge("search_docs", "merge")

Both searches run concurrently; merge runs once, after both finish. The runtime executes in supersteps — everything ready runs together, then state is merged, then the next wave. That is why the reducer question from State and reducers is unavoidable here.

Node signatures

A node takes state and returns a dict. It may also take config:

python 
def node(state: State, config) -> dict:
    user = config["configurable"].get("user_id")
    return {"answer": lookup(user)}

Nodes can be sync or async — use async for anything doing IO, which is most of them. Keep node bodies thin and push logic into ordinary functions you can test without the graph, exactly as with a web handler.

Interview angle 6

  • “How do you express branching?” - a conditional edge whose router is plain Python reading state. Keeping the branch in code rather than in a model call is what makes it deterministic and testable; the model is reserved for decisions that need judgement.
  • “What is Command for?” - returning a state update and the next node together, when the routing decision is a by-product of the work the node just did. Splitting that into a node plus a router means recomputing or stashing the result.
  • “How do you build a loop?” - an edge pointing backwards. An agent is a graph with a cycle and a model in the router — there is nothing else to it.
  • “How do you stop a loop?” - an explicit step counter in state, routed to END when exceeded. The runtime’s recursion limit is a backstop that raises rather than finishing gracefully, so exhaustion should be your decision.
  • “How does parallelism work?” - two edges out of one node run both targets concurrently, and the runtime executes in supersteps: everything ready runs, state merges, next wave. Any key two branches write needs a reducer or it is a conflict.
  • “Why must you compile?” - compile() validates reachability and edge targets and returns the runnable. It is also where the checkpointer attaches, which makes persistence a build-time decision.
Previous
State and reducers