LangGraph Implementation | AI Engineer Course
This chapter provides a hands-on, practical introduction to LangGraph. Instead of jumping straight into complex AI models or LLMs, we will build a fundamental "Hello World" application in LangGraph using a simple mathematical workflow. You will learn how to structure states, define nodes, route execution using conditional edges, and compile your first graph.
1. Demystifying LangGraph: Framework vs. Rocket Science
Many developers view LangGraph as an intimidating, highly complex AI technology. In reality, LangGraph is simply a framework designed to make complex, stateful workflows easier to manage.
Conceptually, any program built with LangGraph can also be written using standard Python if-else blocks, loops, or object-oriented structures. However, as your application grows to include dozens of decision paths, loops, and state updates, writing raw code leads to unmaintainable "spaghetti code." LangGraph simplifies this by providing a structured way to translate flowcharts directly into clean, modular code.
2. The Core Mental Model: The Flowchart Analogy
To understand LangGraph, think back to basic computer science flowcharts. A standard flowchart consists of:
- Start Point: Where the program execution begins.
- Processing Blocks (Rectangles): Where actions or calculations take place.
- Decision Diamonds: Where conditions are evaluated (e.g., "Is ?"), branching into "Yes" or "No" paths.
- End Point: Where the program terminates.
LangGraph is a programmatic implementation of this exact flowchart model. It allows you to define these blocks as code and connect them dynamically.
3. The Three Pillars of LangGraph
Every LangGraph application is built upon three fundamental pillars: State, Nodes, and Edges.
Real-Life Analogy: The Software Engineering Office
Imagine an office with five specialized engineers working on a project (Backend, Frontend, Database, Payment Gateway, and Login System).
- The Whiteboard (State): In the center of the office is a giant whiteboard. When the Database Engineer finishes their task, they write
Database: Readyon the board. The Frontend Engineer reads this update and knows they can now start their work. The whiteboard serves as the single source of truth. - The Engineers (Nodes): Each engineer is a "node." They perform work and update the whiteboard.
- The Tea Server (Helper Function): There is also an office assistant who serves tea. While this assistant is doing work, they do not read or write to the project whiteboard. Therefore, they are not a node. In programming, you may have helper functions, but if they do not directly interact with or modify the State, they are not considered LangGraph nodes.
4. Problem Statement: The Number Doubling Workflow
To master the mechanics of LangGraph without the distraction of LLM APIs, we will implement a simple algorithmic problem:
- Input: An initial integer (e.g.,
5). - Condition: If the number is strictly less than
100, double it. - Loop: Keep doubling the number as long as it remains under
100. - Exit: As soon as the number reaches or exceeds
100, stop execution and print the final value.
Trace of Execution for Input 5:
5. Designing the State (The Shared Whiteboard)
The State represents the data that changes over time. In our workflow, the only variable changing is our numeric value.
In Python, we define this state by subclassing TypedDict from the typing module. This specifies the keys and data types that our whiteboard will track.
from typing import TypedDict
class State(TypedDict):
number: int
Note: You can track as many variables as needed on your whiteboard (e.g.,
name: str,email: str,attempts: int). However, nodes can only update keys that have been explicitly defined in this State class.
6. Implementing Nodes (The Workers)
Nodes are standard Python functions. To integrate seamlessly with LangGraph, they must adhere to two strict rules:
- They must accept the current
Stateas their primary argument. - They must return a dictionary containing the updated state keys.
We will define two nodes: double and finish.
def double(state: State) -> dict:
# 1. Read the current value from the whiteboard
current_number = state["number"]
# 2. Perform the processing logic
new_number = current_number * 2
print(f"[Node: double] Doubled {current_number} to {new_number}")
# 3. Return the updated value to write back to the whiteboard
return {"number": new_number}
def finish(state: State) -> dict:
# Read and print the final value
current_number = state["number"]
print(f"[Node: finish] Workflow complete! Final value is: {current_number}")
# No state changes are made, we return the state as-is
return {"number": current_number}
7. Routing with Conditional Edges (The Decision Maker)
To decide whether to loop back and double the number again or route to the finish node, we need a decision-making function.
This function acts like the "tea server" helper function—it reads the state to make a routing decision, but it does not modify the state. Therefore, it returns a routing string rather than a state dictionary.
def decision(state: State) -> str:
# Evaluate the condition
if state["number"] < 100:
return "double" # Route back to the double node
else:
return "finish" # Route to the finish node
8. Assembling and Compiling the Graph
With our State, Nodes, and routing logic defined, we can now assemble them into a cohesive graph using StateGraph.
Step-by-Step Assembly Workflow:
The Assembly Code:
from langgraph.graph import StateGraph, END
# 1. Initialize the graph builder with our State schema
builder = StateGraph(State)
# 2. Register our nodes with unique string identifiers
builder.add_node("double", double)
builder.add_node("finish", finish)
# 3. Define where the execution starts
builder.set_entry_point("double")
# 4. Configure conditional routing out of the "double" node
builder.add_conditional_edges(
"double", # Source node
decision, # Routing function
{
"double": "double", # If decision returns "double", route to "double" node
"finish": "finish" # If decision returns "finish", route to "finish" node
}
)
# 5. Configure direct routing out of the "finish" node to terminate the graph
builder.add_edge("finish", END)
# 6. Compile the graph into an executable runner
graph = builder.compile()
9. Complete Code Walkthrough
Below is the complete, self-contained Python script. You can run this locally or in a Jupyter Notebook / Google Colab environment.
from typing import TypedDict
from langgraph.graph import StateGraph, END
# ==========================================
# 1. STATE DEFINITION
# ==========================================
class State(TypedDict):
number: int
# ==========================================
# 2. NODE DEFINITIONS
# ==========================================
def double(state: State) -> dict:
current_number = state["number"]
new_number = current_number * 2
print(f"[Node: double] Doubled {current_number} to {new_number}")
return {"number": new_number}
def finish(state: State) -> dict:
current_number = state["number"]
print(f"[Node: finish] Workflow complete! Final value is: {current_number}")
return {"number": current_number}
# ==========================================
# 3. ROUTING FUNCTION
# ==========================================
def decision(state: State) -> str:
if state["number"] < 100:
return "double"
else:
return "finish"
# ==========================================
# 4. GRAPH ASSEMBLY & COMPILATION
# ==========================================
builder = StateGraph(State)
# Add nodes
builder.add_node("double", double)
builder.add_node("finish", finish)
# Set entry point
builder.set_entry_point("double")
# Add conditional routing
builder.add_conditional_edges(
"double",
decision,
{
"double": "double",
"finish": "finish"
}
)
# Add termination edge
builder.add_edge("finish", END)
# Compile
graph = builder.compile()
# ==========================================
# 5. EXECUTION
# ==========================================
if __name__ == "__main__":
initial_input = {"number": 5}
print(f"Starting workflow with input: {initial_input}\n")
# Invoke the graph with our initial state
result = graph.invoke(initial_input)
print(f"\nFinal State returned by Graph: {result}")
Expected Output:
Starting workflow with input: {'number': 5}
[Node: double] Doubled 5 to 10
[Node: double] Doubled 10 to 20
[Node: double] Doubled 20 to 40
[Node: double] Doubled 40 to 80
[Node: double] Doubled 80 to 160
[Node: finish] Workflow complete! Final value is: 160
Final State returned by Graph: {'number': 160}
10. Real-World Pitfalls and Debugging Workflows
When implementing LangGraph in production or local environments, developers frequently encounter several common issues:
1. The "State Key Mismatch" Error
- The Problem: Returning a key from a node that is not defined in your
TypedDictState class. - The Consequence: LangGraph will raise a validation error during execution.
- The Fix: Ensure every dictionary returned by your nodes matches the keys defined in your State class. If you need to store temporary variables, add them to your
TypedDictdefinition first.
2. Syntax Confusion: add_conditional_edges vs. add_edge
- The Problem: Developers often confuse direct routing with conditional routing.
- The Rule:
- Use
add_edge(source, destination)when execution always flows from Node A to Node B without any decision-making. - Use
add_conditional_edges(source, path_function, path_map)when execution routing depends on the output of a decision function.
- Use
3. Google Colab Environment Quirks
- The Problem: When running LangGraph in Google Colab, you may encounter package installation errors, missing dependencies, or copy-paste formatting issues.
- The Fix: Ensure you install the correct packages at the start of your notebook:
If execution hangs or behaves unpredictably, restart your Colab runtime (BASH
!pip install -U langgraph langchain-coreRuntime > Restart session) to clear cached states.
Summary and Key Takeaways
- LangGraph is a workflow framework, not a complex mathematical engine. It translates visual flowcharts into clean, maintainable, stateful code.
- State acts as a shared whiteboard that stores all persistent data. It is defined using Python's
TypedDict. - Nodes are the functional workers of your graph. They must accept the current state as an argument and return a dictionary containing state updates.
- Helper functions are not nodes if they do not directly read or modify the state whiteboard (e.g., routing functions).
- Edges connect your nodes. They can be direct (
add_edge) or conditional (add_conditional_edges). - Compilation is mandatory. You must call
builder.compile()to turn your blueprint into an executable graph object. - Execution begins with
.invoke(), which accepts a dictionary containing the initial values for your state.