Graph Fundamentals in Practice
Explore the key principles of LangGraph by building a branching AI workflow. Learn how to define state schemas, write node functions, connect nodes with edges, compile workflows, and diagnose common issues to effectively manage state and routing in graph-based AI agents.
We'll cover the following...
- The seven-step build pattern
- What we are building
- Step 1: define the state
- Step 2: write the node functions
- Steps 3 and 4: create the builder and register nodes
- Step 5: connect nodes with edges
- Step 6: compile
- Step 7: invoke and read the result
- How state evolves between nodes
- Unconditional versus conditional edges
- Common mistakes
- Complete executable code
- Exercise
- Solution
- LangGraph terms from this lesson
In the previous lesson, we built a working branching workflow. We defined state, wrote nodes, connected edges, and called app.invoke(...). It ran, it routed correctly, it returned results.
But we moved fast. A lot happened under the hood that we did not look at closely.
What does compile() actually do? What happens to state between nodes? When a node returns {"route": "answer"}, where does that update go, and how does the next node see it? What is the difference between add_edge and add_conditional_edges at a mechanics level?
These are practical questions. The moment something in your graph behaves unexpectedly, a node reads a stale value, a route goes to the wrong branch, a state field is missing, understanding the mechanics is what lets you diagnose it quickly instead of guessing.
This lesson fills in those gaps. We will build a richer workflow, slow down at each step, and look at what the LangGraph API is doing.
The seven-step build pattern
Every LangGraph workflow follows the same construction sequence. The nodes and edges grow more complex as the lessons progress, but this structure never changes. The table below outlines the seven steps to build any LangGraph workflow.
Step | What you do | API |
1 | Define the state schema |
|
2 | Write node functions |
|
3 | Create the graph builder |
|
4 | Register nodes |
|
5 | Connect nodes with edges |
|
6 | Compile the graph |
|
7 | Invoke with initial state |
|
We will go through each step carefully below, using a practical FAQ assistant as our working example.
What we are building
The workflow in this lesson handles user questions in four stages. Account-related questions pass through a retrieval step before reaching the drafting node, while general questions skip straight to drafting. Every response then passes through a quality gate before the workflow ends.
Step 1: define the state
State is the object that flows through the graph from node to node. Every node reads from it. Every node returns a partial update that LangGraph merges back into it. By the time execution reaches END, state holds the accumulated output of the entire workflow.
For our FAQ assistant, we need to carry the original question, the route decision, any retrieved context, the drafted response, and a quality result.
from typing import TypedDictclass FAQState(TypedDict):user_question: strroute: strcontext: strdraft_response: strquality_passed: bool
Line 1: Standard library import. No LangGraph dependency here.
Line 3–9: Five fields covering the full data lifecycle of one request.
A few things worth noting about this schema. Every field that any node might read must be declared here: you cannot add fields at runtime. But nodes do not have to update every field. A node that only touches draft_response returns {"draft_response": "..."} and leaves everything else unchanged. LangGraph merges the partial update into the full state, so other fields keep their current values.
Step 2: write the node functions
Each node is a plain Python function. It receives the current state and returns a dictionary. The dictionary is a partial update containing only the fields this node is responsible for changing.
The classifier node reads the user’s question and writes a routing decision:
def classify_message(state: FAQState) -> dict:text = state["user_question"].lower()if any(word in text for word in ["account", "bill", "charge", "invoice", "subscription"]):return {"route": "account"}return {"route": "general"}
Line 1: Takes state, returns a partial update. This signature is the same for every node.
Line 2: Reads the question from state and lowercases it for matching.
Line 3–4: Returns
route=accountfor billing and account-related topics.Line 5: Everything else is a general query.
The retrieval node only runs for account queries. It simulates fetching relevant account context. In a real system, this would query a database or vector store.
def retrieve_context(state: FAQState) -> dict:question = state["user_question"]# In a real workflow, this would query an account database or vector store.context = (f"Account context for query '{question}': ""Customer is on the Pro plan. Last invoice was paid on time. ""No outstanding charges.")return {"context": context}
Line 1: Node receives state. The route decision has already been written by the previous node.
Line 2: Reads the original question to tailor the context retrieval.
Line 3–8: Produces a context string. In production this would be a real lookup.
Line 9: Returns only the
contextfield. The routing and question fields are untouched.
The drafting node calls the LLM to produce a response. It reads both user_question and context from state. Context is available if the retrieval node ran, or empty if it did not.
from groq import Groqdef draft_response(state: FAQState) -> dict:api_key = "{{GROQ_API_KEY}}"client = Groq(api_key=api_key)context_block = ""if state["context"]:context_block = f"\n\nRelevant context:\n{state['context']}"prompt = ("You are a helpful FAQ assistant. ""Answer the following question clearly in 3-5 lines."f"{context_block}\n\n"f"Question: {state['user_question']}")response = client.chat.completions.create(model="llama-3.1-8b-instant",messages=[{"role": "user", "content": prompt}],)return {"draft_response": response.choices[0].message.content}
Line 1: Imports the Groq library.
Line 3–4: Configures the model using the course API key.
Line 6–8: Builds an optional context block. If the retrieval node ran, its output is included. If not,
contextis an empty string and the block is omitted, and the prompt adapts to whatever is in state.Line 10–15: Builds the full prompt.
Line 16–19: Calls Llama, a large language model, and writes the response text to state.
The quality gate node checks the draft and sets a boolean flag. This keeps quality evaluation logic in one dedicated node rather than mixing it into the drafting node.
def quality_gate(state: FAQState) -> dict:draft = state["draft_response"]passed = len(draft.strip()) >= 40return {"quality_passed": passed}
Line 1–2: Reads the draft response from state.
Line 3: Applies a simple length check as a quality proxy.
Line 4: Writes the result as a boolean. A downstream node or edge can read this to decide what to do next.
Steps 3 and 4: create the builder and register nodes
StateGraph is the core LangGraph class for building a workflow. We create it by passing our state schema as the argument. This tells LangGraph what the state object looks like and which fields nodes are allowed to update.
from langgraph.graph import END, START, StateGraphbuilder = StateGraph(FAQState)
Line 1: Imports the three things we always need:
StateGraph,START, andEND.Line 3: Creates the graph builder. The schema we pass here governs every node in this graph.
Now we register our four nodes. The first argument is the name LangGraph will use to identify this node in edges. The second is the Python function.
builder.add_node("classify_message", classify_message)builder.add_node("retrieve_context", retrieve_context)builder.add_node("draft_response", draft_response)builder.add_node("quality_gate", quality_gate)
Line 1–4: Each call registers one node. The string name is what you use in
add_edgecalls. The function is what LangGraph calls when that node executes.
The string name and the function name do not have to match, but keeping them the same makes the graph far easier to read. When you see "draft_response" in an edge, you want to immediately know which function handles it.
Step 5: connect nodes with edges
Edges define what runs next. LangGraph has two kinds: unconditional and conditional.
An unconditional edge always goes to the same next node, regardless of state:
builder.add_edge("retrieve_context", "draft_response")
After retrieve_context runs, execution always moves to draft_response. There is no decision involved. The edge is fixed.
A conditional edge calls a routing function after a node runs and follows the node name that function returns:
from typing import Literaldef route_after_classify(state: FAQState) -> Literal["retrieve_context", "draft_response"]:if state["route"] == "account":return "retrieve_context"return "draft_response"builder.add_conditional_edges("classify_message", route_after_classify)
Line 3: The routing function takes state and returns the name of the next node as a string.
Line 4–6: Reads the
routefield written byclassify_messageand maps it to a node name.Line 8: Registers this as the edge out of
classify_message. After that node runs, LangGraph callsroute_after_classifywith the current state to decide where to go next.
The Literal type annotation on the routing function tells LangGraph exactly which node names are valid return values. During compilation, LangGraph uses this to verify that every possible return value corresponds to a registered node. If you return a name that does not exist, compilation fails with a clear error, which is far better than a silent runtime failure.
Now we connect the remaining edges:
builder.add_edge(START, "classify_message")builder.add_edge("draft_response", "quality_gate")builder.add_edge("quality_gate", END)
Line 1: Connects the entry point.
STARTis a built-in constant: whenapp.invoke(...)is called, execution begins at whatever nodeSTARTconnects to.Line 2: After drafting, always run the quality gate.
Line 3: After quality evaluation, the workflow is done.
Step 6: compile
Compilation is the step that turns the builder into a runnable application, and it does more than check syntax.
app = builder.compile()
When we call compile(), LangGraph performs several checks:
Every node registered with
add_nodeis reachable fromSTART.Every node name referenced in
add_edgeandadd_conditional_edgesexists.Every possible return value of every routing function is a registered node name.
There is at least one path from
STARTtoEND.
If any of these checks fail, compile() raises an error before any execution happens. This is a significant advantage over discovering structural problems at runtime, possibly halfway through a multi-step workflow.
The object returned by compile() is what we call app. It has one main method we care about right now: invoke.
Step 7: invoke and read the result
invoke runs the compiled workflow. We pass it an initial state dictionary with a value for every field declared in our schema.
result = app.invoke({"user_question": "How do I change my notification settings?","route": "","context": "","draft_response": "","quality_passed": False,})
Line 1:
invokeruns the full workflow synchronously and returns the final state.Line 2–7: Initial state values.
route,context, anddraft_responsestart empty because those fields are written by nodes during execution.quality_passedstarts asFalseand will be updated by the quality gate.
After execution, result is a plain dictionary containing the fully updated state. Every field reflects what the nodes wrote into it.
for key, value in result.items():print(f"{key}: {value}")
Printing every field gives us a complete picture of what happened during execution:
user_question: How do I change my notification settings?route: generalcontext:draft_response: To change your notification settings, go to your account...quality_passed: True
Reading the full state is a habit worth forming early. The route tells us which path execution took. The context field shows whether retrieval ran. The quality_passed flag shows whether the gate passed. Every field in state is a record of a decision made by a node during execution.
How state evolves between nodes
It helps to visualise exactly what state looks like as it moves through the graph. Here is the state dictionary at each point in a general-question execution:
After START (initial state)───────────────────────────user_question : "How do I change my notification settings?"route : ""context : ""draft_response: ""quality_passed: FalseAfter classify_message───────────────────────────user_question : "How do I change my notification settings?"route : "general" ← written by classify_messagecontext : ""draft_response: ""quality_passed: FalseAfter draft_response (retrieve_context was skipped)───────────────────────────user_question : "How do I change my notification settings?"route : "general"context : ""draft_response: "To change your..." ← written by draft_responsequality_passed: FalseAfter quality_gate───────────────────────────user_question : "How do I change my notification settings?"route : "general"context : ""draft_response: "To change your..."quality_passed: True ← written by quality_gate
Each node wrote exactly the fields it owned. Everything else stayed as it was. This is the partial update principle in action: classify_message returns {"route": "general"} and does not touch context, draft_response, or quality_passed.
Unconditional versus conditional edges
The distinction between the two edge types is worth being explicit about, because it governs when state is read for routing decisions.
An unconditional edge is recorded at compile time. When you write
builder.add_edge("draft_response", "quality_gate"), LangGraph records this connection during compilation. At runtime it does not evaluate anything: it just follows the fixed connection.A conditional edge is evaluated at runtime. When
classify_messagefinishes, LangGraph callsroute_after_classify(state)with the current state at that moment, and follows whichever node name comes back. The decision is live, based on what the previous node wrote.
The table below compares unconditional and conditional edges, explaining when they are resolved, how they are added, when to use them, and what can go wrong in each case.
Unconditional | Conditional | |
When is it resolved? | At compile time | At runtime, after the source node runs |
How do you add it? |
|
|
When do you use it? | Next step is always the same | Next step depends on something in state |
What can go wrong? | Fixed at compile time | Routing function returns an unregistered node name |
Common mistakes
These are the three errors that appear most often when building a first LangGraph workflow.
Forgetting to connect a node to END.Every execution path must terminate. If you register a node and wire it into the graph but never connect it to
END, the graph either raises a compile error or hangs at runtime. The symptom is easy to read: LangGraph will tell you which node has no outgoing edge.Putting routing logic inside a node instead of a routing function.Nodes write to state. They do not decide where execution goes next. If you find yourself trying to return a node name from inside a node function, you are mixing two responsibilities. Extract the routing logic into a separate function and attach it with
add_conditional_edges.
# Don't do this — nodes should not decide their own successordef classify_message(state: FAQState) -> dict:if "account" in state["user_question"].lower():return {"route": "account", "next_node": "retrieve_context"} # wrong patternreturn {"route": "general", "next_node": "draft_response"} # wrong pattern
The correct version keeps the node focused on writing state, and delegates the routing decision to a separate function:
def classify_message(state: FAQState) -> dict:if "account" in state["user_question"].lower():return {"route": "account"}return {"route": "general"}def route_after_classify(state: FAQState) -> Literal["retrieve_context", "draft_response"]:return "retrieve_context" if state["route"] == "account" else "draft_response"builder.add_conditional_edges("classify_message", route_after_classify)
Not initialising all state fields before invoking.The initial state dictionary passed to
invokemust contain a value for every field in the schema. Missing fields do not get default values. LangGraph will raise aKeyErrorwhen the first node tries to read a missing key.
Complete executable code
Here is the complete workflow in one file, combining all the steps above. Run the code and confirm that each question takes the correct route and produces a meaningful response.
Lines 3–6: Imports.
Literalfor routing return types,TypedDictfor the schema,Groqfor the LLM client, and LangGraph primitives.Lines 11–16: State schema with five fields covering the full data lifecycle of one request.
Lines 21–25: Classifier node. Keyword matching, returns only
route.Lines 28–35: Retrieval node. Builds context for account queries. In a real system, this calls a database or vector store.
Lines 38–56: Drafting node. Reads
user_questionand optionalcontext, builds the prompt, calls Groqchat.completions.createwithllama-3.1-8b-instant, and returns onlydraft_response.Lines 59–61: Quality gate. A length check as a simple quality proxy.
Lines 66–71: Routing function. Reads
routefrom state after classification and returns the next node name.Lines 76–90: Graph assembly. Registers four nodes, wires five edges, and compiles.
Lines 95–102: Clean invocation helper with all fields initialised.
Lines 105–121: Main function. Runs two examples (one general, one account-related) and prints the full state for each.
Lines 124–125: Standard Python entry point that runs
main()when the script is executed directly.
The table below shows what the output should look like for each of the two questions included in the script.
Question type | Route | Context | Quality |
General (notification settings) |
| none | passed |
Account-related (unexpected charge) |
| retrieved | passed |
If quality_passed is False for either question, it means LLM returned an unusually short response. Try running again, as it can occasionally happen with very short model outputs.
Exercise
The workflow currently has two routes out of classify_message: general and account. Add a third route for technical support questions. Here is what the new route needs to do:
If the question contains
error,bug,crash, ornot working, classify it astechnical.Add a new node named
technical_triagethat returns a response asking the user to share their error message and platform details.Connect
technical_triagetoEND.Update the routing function so
route=technicalleads totechnical_triage.
Solution
First, extend the classifier to detect technical keywords:
def classify_message(state: FAQState) -> dict:text = state["user_question"].lower()if any(w in text for w in ["error", "bug", "crash", "not working"]):return {"route": "technical"}if any(w in text for w in ["account", "bill", "charge", "invoice", "subscription"]):return {"route": "account"}return {"route": "general"}
Line 3–4: Checks for technical keywords before account keywords. More specific patterns should be matched first.
Add the new node:
def technical_triage(state: FAQState) -> dict:return {"draft_response": ("To help you faster, could you share the exact error message you are seeing ""and which platform or browser you are using?")}
Line 1–7: Returns a focused data-gathering response. A targeted question is more useful than a generated one here.
Update the routing function and register the new node:
def route_after_classify(state: FAQState,) -> Literal["retrieve_context", "draft_response", "technical_triage"]:mapping = {"account": "retrieve_context","technical": "technical_triage","general": "draft_response",}return mapping.get(state["route"], "draft_response")
Line 3:
Literalnow includes the new node name so compilation can validate it.Line 4–8: A mapping dictionary replaces the inline
if/else. It stays cleaner as the number of routes grows.
In build_graph(), add:
builder.add_node("technical_triage", technical_triage)builder.add_edge("technical_triage", END)
Line 1: Registers the node.
Line 2: Connects it to
ENDso this path terminates cleanly.
LangGraph terms from this lesson
These are the new concepts this lesson added to our working vocabulary.
Term | Meaning |
Unconditional edge | A fixed connection between two nodes; the same path is always taken |
Conditional edge | A connection resolved at runtime by a routing function that reads current state |
Routing function | A plain Python function that returns the name of the next node to execute |
Compile | Validates the graph structure and returns a runnable application object |
Partial update | A dict returned by a node that updates only the fields it owns; other fields are unchanged |
We now understand exactly what happens at each step of the LangGraph build process: how state moves between nodes, what compile validates, and how the two edge types work differently. The next lesson focuses entirely on state design: how to structure state schemas so they stay clean and maintainable as the workflow grows.