Compilation and Execution

Capsule overview

You've defined state, nodes, and edges. You have all the ingredients of a graph. But until you compile, you have nothing runnable — it's like having source code you never built. graph.compile() takes your definition and turns it into an object you can run, stream, and visualize.

This capsule covers the three final phases of a graph's lifecycle: compilation (validating and creating the runnable), execution (the three ways to run it: invoke, stream, and batch), and visualization (draw_mermaid_png() as an essential debugging tool). Of the three, visualization is the one you'll most underestimate and the one that will save you the most time. Before debugging code, look at the picture of your graph.

By the end of this capsule, you'll have the full flow: define state → create nodes → connect edges → compile → run → visualize. In the module's project (Chatbot with state and routing), you'll apply this whole cycle to build a system that classifies intent and routes to specialized nodes.

graph.compile(): from definition to runnable

What compile does

compile() takes your StateGraph (the definition) and produces a CompiledGraph (the runnable). During compilation, LangGraph:

1. Validates the structure — Checks that every node referenced in edges exists, that there's a path from START, and that the graph is consistent
2. Creates the runnable — Produces an object that implements LangChain's Runnable interface, which means you can use invoke, stream, batch, and their async versions
3. Freezes the definition — After compiling, you can't add more nodes or edges

Basic compilation

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    step_count: int

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: ChatState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# Compile
graph = graph_builder.compile()
print(type(graph))
# Output: <class 'langgraph.graph.state.CompiledStateGraph'>

From this point on, graph is your runnable object. graph_builder has done its job — it's the blueprint. graph is the finished building.

Common compilation errors

If your definition has problems, compile() catches them before you run:

graph_builder.add_edge(START, "nonexistent_node")  # ← This node doesn't exist
graph = graph_builder.compile()
# ValueError: Node `nonexistent_node` is not present...

Think of compile() as your graph's linter.

graph.invoke(): run it all

invoke() is the most direct mode: you pass an initial state, the graph runs every node in order, and it hands you back the complete final state.

Basic execution

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    step_count: int

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: ChatState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content="What is RAG?")],
    "step_count": 0
})
print(type(result))           # <class 'dict'>
print(result["step_count"])   # 1
print(len(result["messages"]))  # 2 (HumanMessage + AIMessage)
print(result["messages"][-1].content[:100])
# Output: RAG (Retrieval-Augmented Generation) is a technique that combines the retrieval of...

The result is the complete final state

invoke() returns a dictionary with all the state fields after the last node finished. It's the final snapshot of the state, with all the reducers already applied.

Execution with multiple nodes

When the graph has multiple nodes, invoke() runs them in order. The result accumulates everything each node contributed (respecting the state's reducers):

# With a 2-node graph: classify → respond
result = graph.invoke({
    "messages": [HumanMessage(content="How does async/await work in Python?")],
    "steps_completed": []
})
print(f"Steps completed: {result['steps_completed']}")
# Output: ['classify:technical', 'respond']
# Both nodes contributed to the list (operator.add)

graph.stream(): step by step

invoke() waits for everything to finish. stream() shows you each step as it happens — you see which node ran and what changed after each one.

stream_mode="values": the full state after each node

With stream_mode="values", you get the complete state after each node finishes:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    findings: Annotated[list[str], operator.add]
    current_step: str

model = init_chat_model("openai:gpt-4.1-mini")

def search(state: ResearchState) -> dict:
    topic = state["messages"][-1].content
    response = model.invoke(f"Find one key fact about: {topic}. Answer in one sentence.")
    return {
        "findings": [response.content.strip()],
        "current_step": "analyze"
    }

def analyze(state: ResearchState) -> dict:
    return {
        "findings": ["Analysis completed with high confidence"],
        "current_step": "done"
    }

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", END)
graph = graph_builder.compile()

print("=== stream_mode='values' ===")
for step in graph.stream(
    {
        "messages": [HumanMessage(content="LangGraph")],
        "findings": [],
        "current_step": "search"
    },
    stream_mode="values"
):
    print(f"  current_step: {step['current_step']}")
    print(f"  findings count: {len(step['findings'])}")
    print()
# Output:
# === stream_mode='values' ===
#   current_step: search
#   findings count: 0
#
#   current_step: analyze
#   findings count: 1
#
#   current_step: done
#   findings count: 2

Each iteration gives you a complete snapshot of the state. The first emission is the initial state (before any node). The rest are the state after each node.

stream_mode="updates": only each node's changes

With stream_mode="updates", you get only what each node returned — the partial update, not the full state. Using the same graph as above:

# Same graph, different stream_mode
print("=== stream_mode='updates' ===")
for step in graph.stream(
    {
        "messages": [HumanMessage(content="LangGraph")],
        "findings": [],
        "current_step": "search"
    },
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"  [{node_name}] returned: {list(update.keys())}")
# Output:
# === stream_mode='updates' ===
#   [search] returned: ['findings', 'current_step']
#   [analyze] returned: ['findings', 'current_step']

When to use each mode?

For debugging, "updates" is more useful because you see exactly what each node returned. For a user interface, "values" is better because you always have the complete state available.

Batch: multiple inputs through the same graph

If you need to run the same graph with multiple inputs, batch() is more efficient than calling invoke() in a loop. It takes a list of initial states and returns a list of results in the same order:

inputs = [
    {"messages": [HumanMessage(content="What is Python?")], "step_count": 0},
    {"messages": [HumanMessage(content="What is FastAPI?")], "step_count": 0},
    {"messages": [HumanMessage(content="What is LangGraph?")], "step_count": 0},
]

results = graph.batch(inputs)
for i, result in enumerate(results):
    print(f"[{i+1}] {result['messages'][-1].content[:60]}...")
# Output:
# [1] Python is a high-level programming language...
# [2] FastAPI is a modern, fast web framework for APIs...
# [3] LangGraph is a framework for building AI workflows...

batch() processes every input and returns the list of results. Each result has the same structure as what an individual invoke() would return.

Visualization: draw_mermaid_png()

This is where a lot of developers skip the most important step. Before running your graph, before debugging a bug, before adding more nodes — look at the picture.

Why it's essential

draw_mermaid_png() generates a PNG image of the graph showing every node, edge, and the execution flow. It's living documentation that's always in sync with your code. It's not a nice-to-have — it's your main visual debugging tool.

When a conditional edge doesn't behave the way you expect, the image immediately shows you where the flow can go. When a node never runs, the image reveals there's no edge into it. When the graph is more complex than you thought, the image tells you before you lose hours debugging.

Generating and saving the image

Three lines is all you need:

# After graph = graph_builder.compile()
png_data = graph.get_graph().draw_mermaid_png()
with open("my_graph.png", "wb") as f:
    f.write(png_data)
print("Graph saved to my_graph.png")
# Output: Graph saved to my_graph.png

Visualizing with conditional edges

The image gets especially valuable once you have conditional edges. A graph that routes to three specialized nodes (tech_node, creative_node, general_node) will show classify with three outgoing arrows and all three nodes converging on END. Without the image, you'd have to walk through the code in your head to understand the possible routes.

draw_mermaid() for text

If you can't render PNG (e.g., in a terminal), draw_mermaid() generates the Mermaid text representation:

mermaid_text = graph.get_graph().draw_mermaid()
print(mermaid_text)
# Output:
# %%{init: {'flowchart': {'curve': 'linear'}}}%%
# graph TD;
# 	__start__ --> chatbot;
# 	chatbot --> __end__;

You can copy this text and paste it into mermaid.live, GitHub, or Notion to see the diagram.

A working habit: draw first, code second

Make this flow your standard practice:

1. Define the state — what data flows through your graph
2. Draw the graph — on paper or in your head: which nodes, which edges
3. Implement — write the nodes and edges
4. Visualize — draw_mermaid_png() to check the graph is what you expected
5. Run — stream(mode="updates") to see each step
6. Iterate — adjust and repeat from step 3

Compilation options: a preview

compile() accepts optional parameters that unlock advanced functionality. You'll see them in detail in later modules, but it's useful to know they exist:

checkpointer (Module 8: Memory and Persistence)

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

A checkpointer saves the state after each node. This enables persistence, time-travel debugging, and durable execution. You'll go deep on this in Module 8.

interrupt_before / interrupt_after (Module 9: Human-in-the-Loop)

graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["dangerous_action"]
)

Pauses execution before (or after) a specific node to ask for human approval. Requires a checkpointer. You'll see this in Module 9.

For now, it's enough to know that compile() is extensible — the basic version with no arguments is all you need for this module.

Error handling during execution

What happens when a node raises an exception? By default, it propagates and stops the graph:

def step_two(state) -> dict:
    raise ValueError("Something went wrong")  # The graph stops here

try:
    result = graph.invoke({"items": [], "current_step": "one"})
except ValueError as e:
    print(f"Graph failed: {e}")
# Output: Graph failed: Something went wrong

Handling errors inside nodes

The recommended approach is to handle errors inside each node, catching exceptions and recording them in the state:

from typing import TypedDict, Annotated
import operator

class RobustState(TypedDict):
    messages: Annotated[list, operator.add]
    errors: Annotated[list[str], operator.add]
    current_step: str

def safe_search(state: RobustState) -> dict:
    try:
        response = model.invoke(state["messages"])
        return {"messages": [response], "current_step": "process"}
    except Exception as e:
        return {"errors": [f"search_error: {str(e)}"], "current_step": "process"}

Each node handles its own errors. Downstream nodes check state["errors"] to decide how to proceed. In Module 7 (Advanced Flows) you'll see more sophisticated patterns like retry with backoff and circuit breakers.

Complete example: classification, routing, and streaming

Let's bring compilation, streaming, and visualization together in a single example. A graph that classifies the user's intent and routes to specialized nodes:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class AssistantState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    response_type: str

model = init_chat_model("openai:gpt-4.1-mini")

def classify_intent(state: AssistantState) -> dict:
    last_msg = state["messages"][-1].content
    response = model.invoke(
        f"Classify in one word: technical, creative, or general.\n"
        f"Question: {last_msg}\nAnswer with the category ONLY:"
    )
    return {"intent": response.content.strip().lower()}

def route_by_intent(state: AssistantState) -> Literal["tech_respond", "creative_respond", "general_respond"]:
    if "technical" in state["intent"]:
        return "tech_respond"
    elif "creative" in state["intent"]:
        return "creative_respond"
    return "general_respond"

def tech_respond(state: AssistantState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Technical expert. Two sentences max."},
        *state["messages"]
    ])
    return {"messages": [response], "response_type": "technical"}

def creative_respond(state: AssistantState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Creative writer. Two sentences max."},
        *state["messages"]
    ])
    return {"messages": [response], "response_type": "creative"}

def general_respond(state: AssistantState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Friendly, direct assistant. Two sentences max."},
        *state["messages"]
    ])
    return {"messages": [response], "response_type": "general"}

graph_builder = StateGraph(AssistantState)
graph_builder.add_node("classify", classify_intent)
graph_builder.add_node("tech_respond", tech_respond)
graph_builder.add_node("creative_respond", creative_respond)
graph_builder.add_node("general_respond", general_respond)

graph_builder.add_edge(START, "classify")
graph_builder.add_conditional_edges("classify", route_by_intent)
graph_builder.add_edge("tech_respond", END)
graph_builder.add_edge("creative_respond", END)
graph_builder.add_edge("general_respond", END)

graph = graph_builder.compile()

# Visualize
png_data = graph.get_graph().draw_mermaid_png()
with open("assistant_graph.png", "wb") as f:
    f.write(png_data)
print("Graph saved to assistant_graph.png\n")

# Run with streaming
for step in graph.stream(
    {"messages": [HumanMessage(content="How does async/await work in Python?")],
     "intent": "", "response_type": ""},
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"[{node_name}]")
        if "intent" in update:
            print(f"  Intent: {update['intent']}")
        if "messages" in update:
            print(f"  Response: {update['messages'][-1].content[:80]}...")
# Output:
# Graph saved to assistant_graph.png
#
# [classify]
#   Intent: technical
# [tech_respond]
#   Response: async/await in Python lets you write asynchronous code that doesn't block...

The full flow: define state → create nodes → connect edges → compile → visualize → run with streaming.

Troubleshooting

Problem 1: "Node X is not present" at compile time

Symptom: ValueError: Node 'my_node' is not present in the graph. Cause: You referenced a node in an edge that was never added with add_node(). It might be a typo. Fix: Check that every node mentioned in add_edge and add_conditional_edges was registered:

# BAD — typo in the name
graph_builder.add_node("classify", classify_fn)
graph_builder.add_edge(START, "clasify")  # ← missing 's'

# GOOD
graph_builder.add_node("classify", classify_fn)
graph_builder.add_edge(START, "classify")

Problem 2: stream() produces no output

Symptom: The for step in graph.stream(...) loop prints nothing. Cause: The graph runs but no node returns updates for the fields you're watching, or the input is invalid and the graph finishes immediately. Fix: Check that the nodes return dictionaries with at least one field, and that the initial state includes every required field:

# BAD — a node that returns nothing
def my_node(state):
    do_something(state)
    # No dict returned → no update

# GOOD — always returns a dict
def my_node(state):
    do_something(state)
    return {"current_step": "done"}

Problem 3: draw_mermaid_png() fails with a dependency error

Symptom: ImportError or an error when calling draw_mermaid_png(). Cause: The dependency for rendering Mermaid to PNG is missing. Fix: Install the required package:

pip install grandalf

If it still fails, use draw_mermaid() (without _png) to get the text representation and render it at mermaid.live.

Problem 4: invoke() returns an incomplete state

Symptom: Some fields of the result are None or missing. Cause: The nodes don't update those fields and no initial values were provided. Fix: Include initial values for every field when calling invoke():

# BAD — missing fields
result = graph.invoke({"messages": [HumanMessage(content="hello")]})

# GOOD — every field with an initial value
result = graph.invoke({
    "messages": [HumanMessage(content="hello")],
    "intent": "",
    "response_type": "",
    "findings": []
})

Problem 5: batch() returns results in the wrong order

Symptom: The results from batch() don't match the inputs. Cause: This shouldn't happen — batch() preserves order. If you see scrambled results, check you aren't shuffling them afterwards. Fix: Verify with a simple example:

results = graph.batch([input_1, input_2, input_3])
# results[0] corresponds to input_1
# results[1] corresponds to input_2
# results[2] corresponds to input_3

Exercises

Exercise 1: Compile and run a 3-node graph (Easy)

Create a graph with three nodes: greet (adds a greeting), ask (adds a question), and farewell (adds a goodbye). Use a state with messages: Annotated[list[str], operator.add]. Compile, run with invoke, and print all the messages.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ConvoState(TypedDict):
    messages: Annotated[list[str], operator.add]

def greet(state: ConvoState) -> dict:
    return {"messages": ["Hi there! Welcome."]}

def ask(state: ConvoState) -> dict:
    return {"messages": ["How can I help you today?"]}

def farewell(state: ConvoState) -> dict:
    return {"messages": ["See you around! Have a great day."]}

graph_builder = StateGraph(ConvoState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask)
graph_builder.add_node("farewell", farewell)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", "farewell")
graph_builder.add_edge("farewell", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": []})

for msg in result["messages"]:
    print(f"  → {msg}")
# Output:
#   → Hi there! Welcome.
#   → How can I help you today?
#   → See you around! Have a great day.

Exercise 2: Compare stream_mode values vs updates (Easy)

Using the same graph from exercise 1, run it twice: once with stream_mode="values" and once with stream_mode="updates". Print what you get on each iteration and describe the difference.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ConvoState(TypedDict):
    messages: Annotated[list[str], operator.add]

def greet(state: ConvoState) -> dict:
    return {"messages": ["Hi there!"]}

def ask(state: ConvoState) -> dict:
    return {"messages": ["How are you?"]}

graph_builder = StateGraph(ConvoState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", END)
graph = graph_builder.compile()

print("=== stream_mode='values' ===")
for i, step in enumerate(graph.stream({"messages": []}, stream_mode="values")):
    print(f"  Step {i}: messages = {step['messages']}")

print()

print("=== stream_mode='updates' ===")
for i, step in enumerate(graph.stream({"messages": []}, stream_mode="updates")):
    for node, update in step.items():
        print(f"  Step {i}: [{node}] returned messages = {update.get('messages', 'N/A')}")
# Output:
# === stream_mode='values' ===
#   Step 0: messages = []
#   Step 1: messages = ['Hi there!']
#   Step 2: messages = ['Hi there!', 'How are you?']
#
# === stream_mode='updates' ===
#   Step 0: [greet] returned messages = ['Hi there!']
#   Step 1: [ask] returned messages = ['How are you?']

Exercise 3: Visualize a graph with routing (Medium)

Create a graph that takes a number in the state and routes to one of three nodes: positive_node (if > 0), negative_node (if < 0), or zero_node (if == 0). Generate the graph's PNG image and describe what you see.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from langgraph.graph import StateGraph, START, END

class NumberState(TypedDict):
    value: int
    result: str
    steps: Annotated[list[str], operator.add]

def route_number(state: NumberState) -> Literal["positive", "negative", "zero"]:
    if state["value"] > 0:
        return "positive"
    elif state["value"] < 0:
        return "negative"
    return "zero"

def positive_node(state: NumberState) -> dict:
    return {"result": "Positive number!", "steps": ["positive"]}

def negative_node(state: NumberState) -> dict:
    return {"result": "Negative number", "steps": ["negative"]}

def zero_node(state: NumberState) -> dict:
    return {"result": "It's zero", "steps": ["zero"]}

graph_builder = StateGraph(NumberState)
graph_builder.add_node("positive", positive_node)
graph_builder.add_node("negative", negative_node)
graph_builder.add_node("zero", zero_node)

graph_builder.add_conditional_edges(START, route_number)
graph_builder.add_edge("positive", END)
graph_builder.add_edge("negative", END)
graph_builder.add_edge("zero", END)

graph = graph_builder.compile()

png_data = graph.get_graph().draw_mermaid_png()
with open("number_router.png", "wb") as f:
    f.write(png_data)
print("Graph saved to number_router.png")

for test_value in [42, -7, 0]:
    result = graph.invoke({"value": test_value, "result": "", "steps": []})
    print(f"  value={test_value} → {result['result']} (route: {result['steps']})")
# Output:
# Graph saved to number_router.png
#   value=42 → Positive number! (route: ['positive'])
#   value=-7 → Negative number (route: ['negative'])
#   value=0 → It's zero (route: ['zero'])

Exercise 4: Batch with multiple questions (Medium)

Create a graph that takes a question and generates a one-sentence answer. Use batch() to process 4 different questions at once and print the results with each input's index.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class QAState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    answer: str

model = init_chat_model("openai:gpt-4.1-mini")

def answer_question(state: QAState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Answer in exactly one short sentence."},
        *state["messages"]
    ])
    return {"messages": [response], "answer": response.content.strip()}

graph_builder = StateGraph(QAState)
graph_builder.add_node("answer", answer_question)
graph_builder.add_edge(START, "answer")
graph_builder.add_edge("answer", END)
graph = graph_builder.compile()

questions = ["What is Python?", "What is FastAPI?", "What is LangGraph?", "What is a reducer?"]
inputs = [{"messages": [HumanMessage(content=q)], "answer": ""} for q in questions]
results = graph.batch(inputs)

for i, (q, r) in enumerate(zip(questions, results)):
    print(f"[{i+1}] {q} → {r['answer'][:60]}...")
# Output:
# [1] What is Python? → Python is a high-level programming language...
# [2] What is FastAPI? → FastAPI is a modern, fast web framework...
# [3] What is LangGraph? → LangGraph is a framework for building workflows...
# [4] What is a reducer? → A reducer is a function that defines how to combine...

Exercise 5: Stream updates with node monitoring (Advanced)

Create a 3-node graph (fetch → analyze → report) that uses an LLM. Run it with stream_mode="updates" and print which fields each node updated.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class PipelineState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    data: Annotated[list[str], operator.add]
    current_step: str

model = init_chat_model("openai:gpt-4.1-mini")

def fetch(state: PipelineState) -> dict:
    response = model.invoke("Generate 2 facts about the weather, separated by '|'.")
    items = [x.strip() for x in response.content.split("|")]
    return {"data": items, "current_step": "analyze"}

def analyze(state: PipelineState) -> dict:
    data_text = "\n".join(state["data"])
    response = model.invoke(f"Analyze in one sentence:\n{data_text}")
    return {"data": [f"[analysis] {response.content.strip()}"], "current_step": "report"}

def report(state: PipelineState) -> dict:
    return {"current_step": "done"}

graph_builder = StateGraph(PipelineState)
graph_builder.add_node("fetch", fetch)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("report", report)
graph_builder.add_edge(START, "fetch")
graph_builder.add_edge("fetch", "analyze")
graph_builder.add_edge("analyze", "report")
graph_builder.add_edge("report", END)
graph = graph_builder.compile()

for step in graph.stream(
    {"messages": [], "data": [], "current_step": "fetch"},
    stream_mode="updates"
):
    for node_name, update in step.items():
        data_count = len(update.get("data", []))
        print(f"[{node_name}] → step: {update.get('current_step', '?')}, new data: {data_count}")
# Output:
# [fetch] → step: analyze, new data: 2
# [analyze] → step: report, new data: 1
# [report] → step: done, new data: 0

Exercise 6: Graph with error handling and visualization (Advanced)

Create a 3-node graph (search, validate, format) where search can fail randomly. Implement error handling inside the node (no crashing), generate the PNG visualization, and run it with streaming.

from dotenv import load_dotenv
load_dotenv()

import random
from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    results: Annotated[list[str], operator.add]
    errors: Annotated[list[str], operator.add]
    status: str

model = init_chat_model("openai:gpt-4.1-mini")

def search(state: ResearchState) -> dict:
    try:
        if random.random() < 0.3:
            raise ConnectionError("API unavailable")
        topic = state["messages"][-1].content
        response = model.invoke(f"Find 2 facts about: {topic}. Separate them with '|'.")
        return {"results": [x.strip() for x in response.content.split("|")], "status": "ok"}
    except Exception as e:
        return {"errors": [f"search: {e}"], "results": ["(fallback)"], "status": "error"}

def validate(state: ResearchState) -> dict:
    valid = [r for r in state["results"] if len(r) > 10]
    if not valid:
        return {"errors": ["validate: no valid results"], "status": "failed"}
    return {"status": "validated"}

def format_output(state: ResearchState) -> dict:
    results_text = "\n".join(f"- {r}" for r in state["results"])
    return {"results": [f"REPORT:\n{results_text}"], "status": "done"}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search)
graph_builder.add_node("validate", validate)
graph_builder.add_node("format", format_output)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "validate")
graph_builder.add_edge("validate", "format")
graph_builder.add_edge("format", END)
graph = graph_builder.compile()

png_data = graph.get_graph().draw_mermaid_png()
with open("research_pipeline.png", "wb") as f:
    f.write(png_data)
print("Graph saved to research_pipeline.png\n")

for step in graph.stream(
    {"messages": [HumanMessage(content="AI in medicine")],
     "results": [], "errors": [], "status": "starting"},
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"[{node_name}] status={update.get('status', '?')}")
# Output:
# Graph saved to research_pipeline.png
#
# [search] status=ok
# [validate] status=validated
# [format] status=done

Summary

In this capsule you learned:

• graph.compile() validates your definition and creates a runnable object — it's the step that turns the blueprint into a working system
• graph.invoke() runs the whole graph and returns the final state — ideal for runs where you only care about the result
• graph.stream() shows you each step as it runs — with "values" you see the full state, with "updates" you see only each node's changes
• graph.batch() processes multiple inputs in parallel — more efficient than a loop with invoke()
• draw_mermaid_png() is your main visual debugging tool — before debugging code, look at the picture
• draw_mermaid() generates Mermaid text when you can't render PNG
• Compilation options (checkpointer, interrupt_before) unlock advanced functionality you'll see in later modules
• Error handling happens inside the nodes, not around invoke() — each node is responsible for handling its own errors

Next capsule: create_agent vs StateGraph — when to use the high-level abstraction and when to drop down to LangGraph's full control.

Additional resources

1. LangGraph Quick Start — Official tutorial covering compilation and execution step by step
2. How to run a graph — Guide to invoke, stream, and batch
3. Streaming in LangGraph — Detailed documentation on stream modes
4. Visualization — LangGraph Docs — How to use draw_mermaid_png and the alternatives
5. LangGraph Concepts: Compiling — Conceptual reference on compilation
6. Mermaid Live Editor — Online renderer for pasting draw_mermaid() output
7. Runnable Interface — LangChain — The interface CompiledGraph implements (invoke, stream, batch)
8. Error Handling in LangGraph — Patterns for handling errors in graphs

Module 5 — LangChain & LangGraph: From Chains to Agents