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"
PrincipleDescription
PureDoesn't mutate state, has no side effects
Simpleif/elif/else with a direct return
TestableYou can test it in isolation from the graph
ExhaustiveIt 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

CriterionFixed edgeConditional edge
Predictable flow✅ Always A → BDepends on the state
Dynamic decisions✅ Routing by condition
Simplicity✅ One lineMore code
Use casesPipelines, pre/post-processingClassification, 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:

Piececreate_agentYour manual implementation
Model nodeOpaque ("agent")call_model — visible and editable
Tools nodeOpaque ("tools")run_tools — visible and editable
Decision to continueAutomaticshould_continue — you define it
LoopAutomaticAn explicit edge: run_toolscall_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

Aspectcreate_agentStateGraph + conditional edges
FlowFixed (ReAct loop)Anything you design
DecisionsOnly "are there tool calls?"Whatever you define
LoopsOnly model ↔ toolsAny node can go back to any other
Intermediate nodesYou can't add themValidation, logging, classification
ComplexityLow (3 lines)Medium (routing function + mapping)
ControlMinimalTotal

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
  • START and END are 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_agent has a fixed flow. StateGraph gives you total freedom to design any flowchart
  • draw_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

  1. LangGraph Edges — Conceptual Guide — Official documentation on edges and conditional edges
  2. How to add conditional edges — Branching tutorial with conditional edges
  3. LangGraph Visualization — Visualizing graphs with draw_mermaid_png
  4. How to create a ReAct agent from scratch — Implementing the ReAct loop by hand
  5. create_agent vs custom graphs — When to use each approach
  6. LangGraph Recursion Limit — Controlling infinite loops
  7. StateGraph API Reference — Full StateGraph reference

Module 5 — LangChain & LangGraph: From Chains to Agents