Module 5: Introduction to LangGraph
Edges and Conditional Edges
Capsule overview
In the previous capsule you learned that nodes are functions that transform state. But a node on its own does nothing — it needs to be connected to other nodes to form a flow. Those connections are the edges.
Edges define the path your graph's execution follows. There are two kinds: fixed edges (they always go from node A to node B) and conditional edges (they go to different nodes depending on a decision). Fixed edges are like the arrows on a flowchart. Conditional edges are like the decision diamonds: "is this condition met? Yes → go this way. No → go that way."
This distinction is what makes LangGraph more powerful than create_agent. With create_agent, the only possible flow is the ReAct loop (model → any tool calls? → tools → model). With conditional edges, you define the decisions: classify intent, check quality, decide whether to loop or finish.
Fixed edges: connections that always follow the same path
A fixed edge connects two nodes permanently. Every time node A finishes, execution moves to node B:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def step_a(state: State) -> dict:
return {"messages": [AIMessage(content="Step A completed")]}
def step_b(state: State) -> dict:
return {"messages": [AIMessage(content=f"Step B received: {state['messages'][-1].content}")]}
graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [HumanMessage(content="start")]})
for msg in result["messages"][1:]:
print(msg.content)
# Expected output:
# Step A completed
# Step B received: Step A completed
START and END: the graph's entry and exit
Every graph needs START (the entry point) and END (the exit point):
from langgraph.graph import START, END
graph_builder.add_edge(START, "my_first_node") # Required
graph_builder.add_edge("my_last_node", END) # Required
Without an edge from START, the graph doesn't know where to begin. Without an edge to END, it doesn't know when to stop.
Linear pipeline: A → B → C → END
Connecting nodes in sequence creates a pipeline — the simplest pattern, but a useful one:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, SystemMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
language: str
tone: str
def detect_language(state: State) -> dict:
text = state["messages"][-1].content.lower()
spanish = ["hola", "qué", "cómo", "necesito", "ayuda"]
return {"language": "es" if any(w in text for w in spanish) else "en"}
def detect_tone(state: State) -> dict:
text = state["messages"][-1].content.lower()
if any(w in text for w in ["urgent", "help", "error", "problem"]):
return {"tone": "urgent"}
return {"tone": "neutral"}
def generate_response(state: State) -> dict:
lang = "Spanish" if state.get("language") == "es" else "English"
tone_instruction = "Be direct and solution-oriented." if state.get("tone") == "urgent" else "Be friendly."
model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke(
[SystemMessage(content=f"Respond in {lang}. {tone_instruction}")] + state["messages"]
)
return {"messages": [response]}
graph_builder = StateGraph(State)
graph_builder.add_node("detect_language", detect_language)
graph_builder.add_node("detect_tone", detect_tone)
graph_builder.add_node("generate_response", generate_response)
graph_builder.add_edge(START, "detect_language")
graph_builder.add_edge("detect_language", "detect_tone")
graph_builder.add_edge("detect_tone", "generate_response")
graph_builder.add_edge("generate_response", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content="I need urgent help! My server is down.")],
"language": "", "tone": "",
})
print(f"Language: {result['language']}, Tone: {result['tone']}")
print(result["messages"][-1].content[:80])
# Expected output:
# Language: en, Tone: urgent
# I understand this is urgent. To diagnose why your server is down...
Conditional edges: dynamic decisions
A conditional edge picks its destination based on a routing function:
graph_builder.add_conditional_edges(
"source_node", # From which node
routing_function, # The function that decides
{ # Mapping: return value → destination node
"option_a": "node_a",
"option_b": "node_b",
}
)
The routing function receives the state, evaluates a condition, and returns a string indicating which path to take.
Your first conditional edge
A graph that classifies intent and routes to specialized nodes:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
intent: str
def classifier(state: State) -> dict:
text = state["messages"][-1].content.lower()
if any(w in text for w in ["code", "program", "function", "bug"]):
return {"intent": "code"}
elif any(w in text for w in ["story", "tale", "creative", "write"]):
return {"intent": "creative"}
return {"intent": "qa"}
def route_by_intent(state: State) -> str:
return state["intent"]
def code_expert(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
system = SystemMessage(content="You are an expert programmer. Include code in your answers.")
return {"messages": [model.invoke([system] + state["messages"])]}
def creative_writer(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
system = SystemMessage(content="You are a creative writer. Use expressive language.")
return {"messages": [model.invoke([system] + state["messages"])]}
def qa_assistant(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
system = SystemMessage(content="You are a concise assistant. Answer directly.")
return {"messages": [model.invoke([system] + state["messages"])]}
graph_builder = StateGraph(State)
graph_builder.add_node("classifier", classifier)
graph_builder.add_node("code_expert", code_expert)
graph_builder.add_node("creative_writer", creative_writer)
graph_builder.add_node("qa_assistant", qa_assistant)
graph_builder.add_edge(START, "classifier")
graph_builder.add_conditional_edges(
"classifier",
route_by_intent,
{"code": "code_expert", "creative": "creative_writer", "qa": "qa_assistant"}
)
graph_builder.add_edge("code_expert", END)
graph_builder.add_edge("creative_writer", END)
graph_builder.add_edge("qa_assistant", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content="How do I write a function in Python?")],
"intent": "",
})
print(f"Intent: {result['intent']}")
print(result["messages"][-1].content[:80])
# Expected output:
# Intent: code
# To create a function in Python, use the `def` keyword...
The flowchart shows the decision diamond: classifier → three possible paths → END.
Designing routing functions
The routing function is simple: it takes state, returns a string. It should be pure, testable, and exhaustive:
def route_by_intent(state: State) -> str:
intent = state.get("intent", "qa")
if intent == "code":
return "code"
elif intent == "creative":
return "creative"
return "qa"
# Testable in isolation
assert route_by_intent({"intent": "code", "messages": []}) == "code"
assert route_by_intent({"intent": "unknown", "messages": []}) == "qa"
assert route_by_intent({"messages": []}) == "qa"
| Principle | Description |
|---|---|
| Pure | Doesn't mutate state, has no side effects |
| Simple | if/elif/else with a direct return |
| Testable | You can test it in isolation from the graph |
| Exhaustive | It always has a default else |
The mapping: translating return values into nodes
The third parameter decouples the decision logic from the node names:
graph_builder.add_conditional_edges(
"classifier",
route_by_intent, # Returns "code"
{"code": "code_expert"} # "code" → the "code_expert" node
)
If the function returns a value that isn't in the mapping, you'll get an error. Make sure you map every possible return value.
When to use fixed edges vs conditional edges
| Criterion | Fixed edge | Conditional edge |
|---|---|---|
| Predictable flow | ✅ Always A → B | Depends on the state |
| Dynamic decisions | ❌ | ✅ Routing by condition |
| Simplicity | ✅ One line | More code |
| Use cases | Pipelines, pre/post-processing | Classification, loops, branching |
Rule of thumb: Use fixed edges when the flow is always the same. Use conditional edges when the next step depends on what you discovered earlier.
Creating loops: the ReAct pattern by hand
A conditional edge can point to a previous node, creating a loop. Let's recreate create_agent's loop:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from IPython.display import Image, display
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"Sunny, 22°C in {city}"
@tool
def calculator(expression: str) -> str:
"""Evaluate a math expression."""
return str(eval(expression))
tools_list = [get_weather, calculator]
tool_map = {t.name: t for t in tools_list}
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def call_model(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools_list)
return {"messages": [model_with_tools.invoke(state["messages"])]}
def should_continue(state: State) -> str:
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return "end"
def run_tools(state: State) -> dict:
last = state["messages"][-1]
results = []
for tc in last.tool_calls:
result = tool_map[tc["name"]].invoke(tc["args"])
results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
return {"messages": results}
graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("run_tools", run_tools)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
"call_model",
should_continue,
{"tools": "run_tools", "end": END}
)
graph_builder.add_edge("run_tools", "call_model") # ← The loop
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content="What's the weather in Madrid and what is 25 * 4?")]
})
print(result["messages"][-1].content)
# Expected output: The weather in Madrid is sunny at 22°C. And 25 × 4 = 100.
This is exactly what create_agent does internally, except now you control every piece:
| Piece | create_agent | Your manual implementation |
|---|---|---|
| Model node | Opaque ("agent") | call_model — visible and editable |
| Tools node | Opaque ("tools") | run_tools — visible and editable |
| Decision to continue | Automatic | should_continue — you define it |
| Loop | Automatic | An explicit edge: run_tools → call_model |
Routing to multiple nodes (3+)
Conditional edges can route to any number of destinations:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def router(state: State) -> dict:
return {}
def route_by_content(state: State) -> str:
text = state["messages"][-1].content.lower()
if any(w in text for w in ["code", "program", "bug"]):
return "code_help"
elif any(w in text for w in ["write", "story", "poem"]):
return "creative"
elif any(w in text for w in ["translate", "translation", "spanish"]):
return "translation"
return "general_qa"
def make_specialist(system_prompt: str):
def specialist(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
return {"messages": [model.invoke([SystemMessage(content=system_prompt)] + state["messages"])]}
return specialist
graph_builder = StateGraph(State)
graph_builder.add_node("router", router)
graph_builder.add_node("code_node", make_specialist("You are an expert programmer. Respond in English."))
graph_builder.add_node("creative_node", make_specialist("You are a creative writer. Respond in English."))
graph_builder.add_node("translation_node", make_specialist("You are a professional translator. Respond in English."))
graph_builder.add_node("qa_node", make_specialist("You are a concise assistant. Respond in English."))
graph_builder.add_edge(START, "router")
graph_builder.add_conditional_edges(
"router", route_by_content,
{"code_help": "code_node", "creative": "creative_node",
"translation": "translation_node", "general_qa": "qa_node"}
)
for node in ["code_node", "creative_node", "translation_node", "qa_node"]:
graph_builder.add_edge(node, END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
for msg in ["Fix this bug", "Write a poem", "Translate 'hello'", "Capital of Japan?"]:
result = graph.invoke({"messages": [HumanMessage(content=msg)]})
print(f"'{msg}' → {result['messages'][-1].content[:50]}...")
# Expected output:
# 'Fix this bug' → Sure, I'll need to see the code to help you...
# 'Write a poem' → The moon, a silver beacon on the silent sea...
# 'Translate 'hello'' → "Hello" translates to "Hola" in Spanish...
# 'Capital of Japan?' → The capital of Japan is Tokyo...
The make_specialist factory function avoids duplicating code for similar nodes. The diagram shows 4 routes leaving the router.
Comparison: create_agent's flow vs your own flow
| Aspect | create_agent | StateGraph + conditional edges |
|---|---|---|
| Flow | Fixed (ReAct loop) | Anything you design |
| Decisions | Only "are there tool calls?" | Whatever you define |
| Loops | Only model ↔ tools | Any node can go back to any other |
| Intermediate nodes | You can't add them | Validation, logging, classification |
| Complexity | Low (3 lines) | Medium (routing function + mapping) |
| Control | Minimal | Total |
When to pick which?
- ✅
create_agent: when the ReAct loop is enough (80% of agents) - ✅
StateGraph: when you need dynamic routing, validation, custom loops, or non-linear flows
Troubleshooting
Problem 1: "Routing function returned unexpected value"
Symptom: An error because the function returned a value with no key in the mapping. Fix: Make sure every possible return value is mapped. Add a default:
def route(state: State) -> str:
intent = state.get("intent", "qa")
if intent in ("code", "creative", "qa"):
return intent
return "qa" # Safe default
Problem 2: "Infinite loop — the graph never finishes"
Symptom: It runs indefinitely until recursion_limit stops it.
Fix: The routing function must be able to return a value that leads to END. Add a counter:
def should_retry(state: State) -> str:
if state.get("attempts", 0) >= 3:
return "end"
return "retry"
Problem 3: "Unreachable nodes"
Symptom: A node never runs.
Cause: No edge points to that node.
Fix: Check with draw_mermaid_png(). If a node shows up isolated, it's missing an inbound edge.
Problem 4: "The conditional edge always goes to the same node"
Symptom: No matter the input, it always takes the same path. Fix: Test the routing function in isolation:
assert route({"intent": "code", "messages": []}) == "code"
assert route({"intent": "creative", "messages": []}) == "creative"
Problem 5: "The edge from START is missing"
Symptom: graph.invoke() returns the state unchanged.
Fix: Always add graph_builder.add_edge(START, "first_node") as your first step.
Exercises
Exercise 1: 3-step linear pipeline (Easy)
Create a graph with "input_cleaner" (strips extra spaces), "word_counter" (counts words, saves to state), and "summarizer" (reports the count). Fixed edges. Visualize it.
See solution
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
word_count: int
def input_cleaner(state: State) -> dict:
cleaned = " ".join(state["messages"][-1].content.split())
return {"messages": [HumanMessage(content=cleaned)]}
def word_counter(state: State) -> dict:
return {"word_count": len(state["messages"][-1].content.split())}
def summarizer(state: State) -> dict:
return {"messages": [AIMessage(content=f"Your message has {state['word_count']} words.")]}
graph_builder = StateGraph(State)
graph_builder.add_node("input_cleaner", input_cleaner)
graph_builder.add_node("word_counter", word_counter)
graph_builder.add_node("summarizer", summarizer)
graph_builder.add_edge(START, "input_cleaner")
graph_builder.add_edge("input_cleaner", "word_counter")
graph_builder.add_edge("word_counter", "summarizer")
graph_builder.add_edge("summarizer", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content=" Hello world from LangGraph ")],
"word_count": 0,
})
print(result["messages"][-1].content)
# Expected output: Your message has 4 words.
Explanation: A linear pipeline: clean → count → report. Each node does one thing.
Exercise 2: Conditional edge by language (Easy)
Create a graph with a "checker" node that detects Spanish vs English, and a conditional edge to "spanish_responder" or "english_responder". Visualize it.
See solution
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
detected_lang: str
def checker(state: State) -> dict:
text = state["messages"][-1].content.lower()
spanish = ["hola", "qué", "cómo", "gracias", "necesito", "ayuda"]
return {"detected_lang": "es" if any(w in text for w in spanish) else "en"}
def route_language(state: State) -> str:
return state["detected_lang"]
def spanish_responder(state: State) -> dict:
return {"messages": [AIMessage(content="¡Entendido! Respondo en español.")]}
def english_responder(state: State) -> dict:
return {"messages": [AIMessage(content="Got it! I'll respond in English.")]}
graph_builder = StateGraph(State)
graph_builder.add_node("checker", checker)
graph_builder.add_node("spanish_responder", spanish_responder)
graph_builder.add_node("english_responder", english_responder)
graph_builder.add_edge(START, "checker")
graph_builder.add_conditional_edges(
"checker", route_language,
{"es": "spanish_responder", "en": "english_responder"}
)
graph_builder.add_edge("spanish_responder", END)
graph_builder.add_edge("english_responder", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
for msg in ["Hola, necesito ayuda", "Hello, I need help"]:
result = graph.invoke({"messages": [HumanMessage(content=msg)], "detected_lang": ""})
print(f"'{msg}' → {result['messages'][-1].content}")
# Expected output:
# 'Hola, necesito ayuda' → ¡Entendido! Respondo en español.
# 'Hello, I need help' → Got it! I'll respond in English.
Explanation: The conditional edge routes to two nodes based on the language. The diagram shows the diamond with two paths.
Exercise 3: Loop with an exit condition (Medium)
Create a graph with an "improve" node that bumps quality_score by 0.3 each time, and a conditional edge that stops if it's >= 0.8 or goes back to "improve". Count the iterations. Visualize it.
See solution
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
quality_score: float
iterations: int
def improve(state: State) -> dict:
new_score = min(state.get("quality_score", 0.0) + 0.3, 1.0)
iters = state.get("iterations", 0) + 1
return {
"quality_score": new_score,
"iterations": iters,
"messages": [AIMessage(content=f"Iteration {iters}: score = {new_score:.1f}")],
}
def check_quality(state: State) -> str:
return "done" if state["quality_score"] >= 0.8 else "improve_more"
graph_builder = StateGraph(State)
graph_builder.add_node("improve", improve)
graph_builder.add_edge(START, "improve")
graph_builder.add_conditional_edges(
"improve", check_quality,
{"done": END, "improve_more": "improve"}
)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [], "quality_score": 0.0, "iterations": 0})
print(f"Final score: {result['quality_score']:.1f} | Iterations: {result['iterations']}")
for msg in result["messages"]:
print(f" {msg.content}")
# Expected output:
# Final score: 0.9 | Iterations: 3
# Iteration 1: score = 0.3
# Iteration 2: score = 0.6
# Iteration 3: score = 0.9
Explanation: The diagram shows the circular arrow (the loop) with a conditional exit to END.
Exercise 4: Recreate the ReAct loop by hand (Medium)
Recreate create_agent's loop: a "model" node (LLM with tools), a "should_continue" conditional edge (are there tool calls?), and a "tools" node (runs the tools). Use a search tool. Compare both graphs visually.
See solution
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from IPython.display import Image, display
@tool
def search(query: str) -> str:
"""Search the internet for information."""
return f"Python was created by Guido van Rossum in 1991."
tools_list = [search]
tool_map = {t.name: t for t in tools_list}
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def model_node(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
return {"messages": [model.bind_tools(tools_list).invoke(state["messages"])]}
def should_continue(state: State) -> str:
last = state["messages"][-1]
return "continue" if hasattr(last, "tool_calls") and last.tool_calls else "end"
def tools_node(state: State) -> dict:
results = []
for tc in state["messages"][-1].tool_calls:
output = tool_map[tc["name"]].invoke(tc["args"])
results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
return {"messages": results}
graph_builder = StateGraph(State)
graph_builder.add_node("model", model_node)
graph_builder.add_node("tools", tools_node)
graph_builder.add_edge(START, "model")
graph_builder.add_conditional_edges("model", should_continue, {"continue": "tools", "end": END})
graph_builder.add_edge("tools", "model")
my_graph = graph_builder.compile()
print("=== My graph ===")
display(Image(my_graph.get_graph().draw_mermaid_png()))
result = my_graph.invoke({"messages": [HumanMessage(content="Who created Python?")]})
print(result["messages"][-1].content)
print("\n=== create_agent ===")
agent = create_agent("openai:gpt-4.1-mini", tools=[search])
display(Image(agent.get_graph().draw_mermaid_png()))
result_agent = agent.invoke({"messages": [("user", "Who created Python?")]})
print(result_agent["messages"][-1].content)
# Expected output: Both answer that Python was created by Guido van Rossum.
Explanation: Both graphs have the same structure. The difference: in your graph you can see and modify should_continue, model_node, and tools_node. In create_agent, everything is automatic but opaque.
Exercise 5: Router with 4 destinations (Advanced)
Create a graph with a "router" node and a conditional edge that routes to 4 specialized nodes (tech, business, science, casual), each with a different system prompt. Test it with 4 messages. Visualize it.
See solution
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
category: str
def classify(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
prompt = SystemMessage(content=(
"Classify into ONE category. Answer with the category ONLY.\n"
"Categories: tech, business, science, casual"
))
result = model.invoke([prompt] + state["messages"])
cat = result.content.strip().lower()
return {"category": cat if cat in {"tech", "business", "science", "casual"} else "casual"}
def route_cat(state: State) -> str:
return state["category"]
def make_node(prompt: str):
def node(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
return {"messages": [model.invoke([SystemMessage(content=prompt)] + state["messages"])]}
return node
graph_builder = StateGraph(State)
graph_builder.add_node("classify", classify)
graph_builder.add_node("tech", make_node("Technology expert. Respond in English."))
graph_builder.add_node("business", make_node("Business consultant. Respond in English."))
graph_builder.add_node("science", make_node("Science communicator. Respond in English."))
graph_builder.add_node("casual", make_node("Conversational friend. Respond in English."))
graph_builder.add_edge(START, "classify")
graph_builder.add_conditional_edges(
"classify", route_cat,
{"tech": "tech", "business": "business", "science": "science", "casual": "casual"}
)
for n in ["tech", "business", "science", "casual"]:
graph_builder.add_edge(n, END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
for msg in ["How does Docker work?", "How do I scale my startup?", "Why is the sky blue?", "How's it going?"]:
result = graph.invoke({"messages": [HumanMessage(content=msg)], "category": ""})
print(f"[{result['category']}] {msg} → {result['messages'][-1].content[:60]}...")
# Expected output:
# [tech] How does Docker work? → Docker is a container platform...
# [business] How do I scale my startup? → To scale your startup...
# [science] Why is the sky blue? → The sky is blue because of the scattering...
# [casual] How's it going? → Pretty good! How about you?...
Explanation: It uses the LLM to classify (smarter than keywords). The make_node factory function avoids duplicating code. The diagram shows 4 routes from the classifier.
Summary
In this capsule you learned:
- Fixed edges (
add_edge("a", "b")) connect nodes permanently — the execution always follows the same path STARTandENDare required special nodes that mark the graph's entry and exit- Conditional edges (
add_conditional_edges) pick the next node based on a routing function that returns a string - The mapping parameter translates the routing function's return values into destination node names
- Routing functions should be pure, simple, testable, and exhaustive
- You can create loops with conditional edges that point back to earlier nodes — that's how the ReAct loop works
- You manually recreated
create_agent's loop: model → any tool calls? → tools → model create_agenthas a fixed flow.StateGraphgives you total freedom to design any flowchartdraw_mermaid_png()shows the decision diamonds and possible paths — it's your map of the workflow
Next capsule: Typed state with TypedDict and Annotated — you'll go deeper into state design, reducers, the prebuilt MessagesState, and complex states for real graphs.
Additional resources
- LangGraph Edges — Conceptual Guide — Official documentation on edges and conditional edges
- How to add conditional edges — Branching tutorial with conditional edges
- LangGraph Visualization — Visualizing graphs with draw_mermaid_png
- How to create a ReAct agent from scratch — Implementing the ReAct loop by hand
- create_agent vs custom graphs — When to use each approach
- LangGraph Recursion Limit — Controlling infinite loops
- StateGraph API Reference — Full StateGraph reference
Module 5 — LangChain & LangGraph: From Chains to Agents