Search⌘ K
AI Features

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.

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

class MyState(TypedDict)

2

Write node functions

def my_node(state) -> dict

3

Create the graph builder

StateGraph(MyState)

4

Register nodes

builder.add_node("name", function)

5

Connect nodes with edges

builder.add_edge(...) or builder.add_conditional_edges(...)

6

Compile the graph

app = builder.compile()

7

Invoke with initial state

app.invoke({...})

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.

FAQ assistant workflow: account queries include a retrieval step before drafting; all responses pass through a quality gate before END
FAQ assistant workflow: account queries include a retrieval step before drafting; all responses pass through a quality gate before END

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 TypedDict
class FAQState(TypedDict):
user_question: str
route: str
context: str
draft_response: str
quality_passed: bool
State schema for the FAQ assistant workflow
  • 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"}
Classify node that sets the route field
  • 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=account for 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}
Retrieval node that adds context to state
  • 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 context field. 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 Groq
def 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}
Drafting node that calls LLM with optional context
  • 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, context is 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()) >= 40
return {"quality_passed": passed}
Quality gate node that validates response length
  • 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, StateGraph
builder = StateGraph(FAQState)
Create the StateGraph builder with our state schema
  • Line 1: Imports the three things we always need: StateGraph, START, and END.

  • 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)
Register all four nodes with the graph builder
  • Line 1–4: Each call registers one node. The string name is what you use in add_edge calls. 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")
Unconditional edge from retrieve_context to 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 Literal
def 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)
Routing function and conditional edge for classify_message
  • Line 3: The routing function takes state and returns the name of the next node as a string.

  • Line 4–6: Reads the route field written by classify_message and maps it to a node name.

  • Line 8: Registers this as the edge out of classify_message. After that node runs, LangGraph calls route_after_classify with 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)
Complete edge wiring from START to END
  • Line 1: Connects the entry point. START is a built-in constant: when app.invoke(...) is called, execution begins at whatever node START connects 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()
Compile the graph into a runnable app

When we call compile(), LangGraph performs several checks:

  • Every node registered with add_node is reachable from START.

  • Every node name referenced in add_edge and add_conditional_edges exists.

  • Every possible return value of every routing function is a registered node name.

  • There is at least one path from START to END.

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,
})
Invoke the graph with a general question
  • Line 1: invoke runs the full workflow synchronously and returns the final state.

  • Line 2–7: Initial state values. route, context, and draft_response start empty because those fields are written by nodes during execution. quality_passed starts as False and 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}")
Inspect the full returned state

Printing every field gives us a complete picture of what happened during execution:

user_question: How do I change my notification settings?
route: general
context:
draft_response: To change your notification settings, go to your account...
quality_passed: True
Expected output for a general question

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: False
After classify_message
───────────────────────────
user_question : "How do I change my notification settings?"
route : "general" ← written by classify_message
context : ""
draft_response: ""
quality_passed: False
After 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_response
quality_passed: False
After 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
State snapshot after each node for a general question

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_message finishes, LangGraph calls route_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?

add_edge(source, destination)

add_conditional_edges(source, fn)

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 successor
def classify_message(state: FAQState) -> dict:
if "account" in state["user_question"].lower():
return {"route": "account", "next_node": "retrieve_context"} # wrong pattern
return {"route": "general", "next_node": "draft_response"} # wrong pattern
Anti-pattern: routing logic inside a node

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)
Correct pattern: routing lives in a dedicated function attached to a conditional edge
  • Not initialising all state fields before invoking.The initial state dictionary passed to invoke must contain a value for every field in the schema. Missing fields do not get default values. LangGraph will raise a KeyError when 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.

Python
"""Four-node FAQ workflow with Groq."""
from typing import Literal, TypedDict
from groq import Groq
from langgraph.graph import END, START, StateGraph
# ── State ─────────────────────────────────────────────────────────────────────
class FAQState(TypedDict):
user_question: str
route: str
context: str
draft_response: str
quality_passed: bool
# ── Nodes ─────────────────────────────────────────────────────────────────────
def classify_message(state: FAQState) -> dict:
text = state["user_question"].lower()
if any(w in text for w in ["account", "bill", "charge", "invoice", "subscription"]):
return {"route": "account"}
return {"route": "general"}
def retrieve_context(state: FAQState) -> dict:
question = state["user_question"]
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}
def 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}
def quality_gate(state: FAQState) -> dict:
passed = len(state["draft_response"].strip()) >= 40
return {"quality_passed": passed}
# ── Routing ───────────────────────────────────────────────────────────────────
def route_after_classify(
state: FAQState,
) -> Literal["retrieve_context", "draft_response"]:
if state["route"] == "account":
return "retrieve_context"
return "draft_response"
# ── Graph ─────────────────────────────────────────────────────────────────────
def build_graph():
builder = StateGraph(FAQState)
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)
builder.add_edge(START, "classify_message")
builder.add_conditional_edges("classify_message", route_after_classify)
builder.add_edge("retrieve_context", "draft_response")
builder.add_edge("draft_response", "quality_gate")
builder.add_edge("quality_gate", END)
return builder.compile()
# ── Main ──────────────────────────────────────────────────────────────────────
def run(app, question: str) -> dict:
return app.invoke({
"user_question": question,
"route": "",
"context": "",
"draft_response": "",
"quality_passed": False,
})
def main() -> None:
app = build_graph()
questions = [
"How do I change my notification settings?",
"Why does my invoice show an unexpected charge?",
]
print("=== FAQ assistant workflow ===\n")
for q in questions:
result = run(app, q)
print(f"Question : {q}")
print(f"Route : {result['route']}")
print(f"Context : {'(retrieved)' if result['context'] else '(none)'}")
print(f"Response : {result['draft_response'][:120]}")
print(f"Quality : {'passed' if result['quality_passed'] else 'failed'}")
print()
if __name__ == "__main__":
main()
  • Lines 3–6: Imports. Literal for routing return types, TypedDict for the schema, Groq for 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_question and optional context, builds the prompt, calls Groq chat.completions.create with llama-3.1-8b-instant, and returns only draft_response.

  • Lines 59–61: Quality gate. A length check as a simple quality proxy.

  • Lines 66–71: Routing function. Reads route from 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)

general

none

passed

Account-related (unexpected charge)

account

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, or not working, classify it as technical.

  • Add a new node named technical_triage that returns a response asking the user to share their error message and platform details.

  • Connect technical_triage to END.

  • Update the routing function so route=technical leads to technical_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"}
Classifier with technical route added
  • 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?"
)
}
Technical triage node
  • 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")
Updated routing function and graph registration
  • Line 3: Literal now 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)
Register and connect the new node
  • Line 1: Registers the node.

  • Line 2: Connects it to END so 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.