Chapter 12
Week 4 Day 2 - LangGraph, the orchestration layer
Week 4 Day 2 - LangGraph, the orchestration layer
Yesterday we met the building blocks. Today we move up to Layer 2 and let LangGraph wire those blocks into a graph that it runs for us, keeping track of state and memory as it goes.
What LangGraph is
LangGraph lets you describe a piece of work as a graph: a handful of nodes, which are just Python functions, joined by edges that decide what runs next. The framework holds a shared object called the state, passes it from node to node, and can save a snapshot at every step so your application can remember and recover.
It's worth appreciating that LangGraph is agnostic to what it's orchestrating - you don't even need an LLM in the picture!
# Imports and environment first, as always
import os
import random
import requests
from typing import Annotated
from typing_extensions import TypedDict
from dotenv import load_dotenv
from IPython.display import Image, display
import gradio as gr
from langchain_openai import ChatOpenAI
from langchain_community.tools import GoogleSerperRun
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
load_dotenv(override=True)The State, and a word about reducers
The state is the object that flows through the graph. We describe it with a TypedDict. Each field updates on its own, and by default a new value simply replaces the old one.
Sometimes we want updates to accumulate instead. A reducer is a small function that says how to combine the old value with the new one. We attach it with Annotated. LangGraph ships the reducer we want for chat, add_messages, which appends new messages to the running list rather than overwriting it.
class State(TypedDict):
messages: Annotated[list, add_messages]Five steps to a graph, with no LLM in sight
Building a graph always follows the same shape: define the state, start a builder, add nodes, add edges, then compile. Here is the whole thing with a node that just makes up a silly sentence, to keep the focus on the mechanics.
nouns = ["Cabbages", "Unicorns", "Toasters", "Penguins", "Bananas", "Eels", "Pickles"]
adjectives = ["outrageous", "smelly", "pedantic", "existential", "sparkly", "haunted"]
def silly_node(state: State) -> dict:
sentence = f"{random.choice(nouns)} are {random.choice(adjectives)}"
return {"messages": [{"role": "assistant", "content": sentence}]}
builder = StateGraph(State)
builder.add_node("silly", silly_node)
builder.add_edge(START, "silly")
builder.add_edge("silly", END)
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))result = graph.invoke({"messages": [{"role": "user", "content": "say something"}]})
print(result["messages"][-1].content)That is the entire pattern, and it shows that a graph is really just Python functions joined together. Now we swap the silly node for one that calls a model.
llm = ChatOpenAI(model="gpt-5.4-mini")
def chatbot_node(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))result = graph.invoke({"messages": [{"role": "user", "content": "What is a directed graph, in one sentence?"}]})
print(result["messages"][-1].content)Adding tools: a node, a condition, and a loop
To let the model use tools, we add two prebuilt pieces. ToolNode is a node that runs whatever tools the model asked for. tools_condition is a ready made router that sends the flow to the tools node when the model wants a tool, and to the end when it is finished.
The tools themselves come from two different places, quite deliberately. The search tool comes off the shelf: langchain-community ships ready-made tools for hundreds of services, and GoogleSerperRun wraps the same Serper API you already have a key for. The push notification tool we write ourselves with the @tool decorator, exactly as in lab 1. Notice that both kinds plug into the graph in exactly the same way.
By the way, you may have noticed a deprecation warning when we ran the imports. The langchain-community package is gradually being wound down in favor of standalone packages, one per integration, the same way langchain-openai ships on its own. It still works well and you will meet it in tutorials all over the web, so it is worth knowing. If you would prefer a fully supported alternative, the langchain-tavily package offers a TavilySearch tool that does the same job with a free API key from tavily.com.
We bind the tools to the model so it knows they exist, then wire a loop: the chatbot decides, the tools run, and we come back to the chatbot so it can see the results and continue.
search = GoogleSerperRun(api_wrapper=GoogleSerperAPIWrapper())
@tool
def send_push_notification(text: str) -> str:
"""Send a short push notification to the user's phone."""
requests.post(
"https://api.pushover.net/1/messages.json",
data={"token": os.getenv("PUSHOVER_TOKEN"), "user": os.getenv("PUSHOVER_USER"), "message": text},
)
return "Notification sent"
tools = [search, send_push_notification]
llm_with_tools = llm.bind_tools(tools)def chatbot_node(state: State) -> dict:
return {"messages": [llm_with_tools.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", tools_condition)
builder.add_edge("tools", "chatbot")
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))result = graph.invoke({"messages": [{"role": "user", "content": "Use your search tool to tell me the ingredients in banoffee pie and send me a push notification."}]})
print(result["messages"][-1].content)A second LLM node: the translator
Nodes do not have to be part of the tool loop. Any function can be a node, including one that makes its own LLM call. To prove it, we will add a translator that writes a Spanish version of the final answer into the state.
Two things are new here. First, the state gains a second field, spanish. It has no reducer, so each run simply overwrites it, which is just what we want. Second, until now tools_condition has sent the finished conversation straight to END. By passing a mapping as the third argument, we send that branch through the translator instead, and the translator ends the run.
Look closely at the rendered graph this time. There are two nodes that call an LLM: the chatbot, which has tools and sits in a loop, and the translator, which makes one plain call on the way out. And since the translator runs after the chatbot has finished, it gets a super-step of its own.
class State(TypedDict):
messages: Annotated[list, add_messages]
spanish: str
def translator_node(state: State) -> dict:
last = state["messages"][-1].content
prompt = f"Translate this into Spanish, replying with the translation only:\n\n{last}"
return {"spanish": llm.invoke(prompt).content}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_node("tools", ToolNode(tools))
builder.add_node("translator", translator_node)
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", tools_condition, {"tools": "tools", END:"translator"})
builder.add_edge("tools", "chatbot")
builder.add_edge("translator", END)
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))result = graph.invoke({"messages": [{"role": "user", "content": "In one sentence, what is special about a banana?"}]})
print(result["messages"][-1].content)
print(result["spanish"])# Now load the env again if you added LangSmith variables
load_dotenv(override=True)Memory: why the graph forgets, and how to fix it
Try chatting twice with the graph above and it will not remember the first exchange. That feels odd, because the state was carefully passing messages along. The reason is that one call to invoke is one run of the graph: a sequence of super-steps, one for each layer of nodes that executes (the LLM call is one super-step; the tool calls it requests run together in the next). The reducer accumulates messages within that run, but the next invoke starts fresh.
To remember across calls we add a checkpointer. It saves a snapshot of the state after every super-step, filed under a thread_id. Give it the same thread_id next time and the saved history is loaded before your new message is added.
memory = MemorySaver()
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_node("tools", ToolNode(tools))
builder.add_node("translator", translator_node)
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", tools_condition, {"tools": "tools", END: "translator"})
builder.add_edge("translator", END)
builder.add_edge("tools", "chatbot")
graph = builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "conversation-1"}}
graph.invoke({"messages": [{"role": "user", "content": "Hi, my name is Ed."}]}, config)
second = graph.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, config)
print(second["messages"][-1].content)
print(second["spanish"])Persisting to SQLite
MemorySaver keeps everything in memory, which vanishes when the program stops. For something durable we swap in the SQLite checkpointer, which writes to a database file. The same graph code now survives restarts. A file called memory.db appears next to this notebook, and you can delete it whenever you like.
with SqliteSaver.from_conn_string("memory.db") as sql_memory:
durable_graph = builder.compile(checkpointer=sql_memory)
sql_config = {"configurable": {"thread_id": "conversation-2"}}
durable_graph.invoke({"messages": [{"role": "user", "content": "Remember that my favorite color is orange."}]}, sql_config)
reply = durable_graph.invoke({"messages": [{"role": "user", "content": "What is my favorite color?"}]}, sql_config)
print(reply["messages"][-1].content)
print(reply["spanish"])Looking inside the state, and travelling back in time
Because the checkpointer saves a snapshot at every super-step, you can inspect the current state and walk the whole history. Each entry carries a checkpoint_id, and by putting one of those into the config you can rerun the graph from that exact moment. This is what lets LangGraph applications recover and branch from any earlier point.
snapshot = graph.get_state(config)
messages = snapshot.values["messages"]
print("Messages stored so far:", len(messages))
for message in messages:
print(message.content)
history = list(graph.get_state_history(config))
print("Number of saved checkpoints:", len(history))for h in reversed(history): # history is newest first; reversed reads as a story
m = h.metadata
queued = ", ".join(t.name for t in h.tasks) or "(run complete)"
print(f"step {m['step']:>2} {m['source']:<5} messages={len(h.values.get('messages', []))} about to run: {queued}")# Time travel: take a checkpoint from earlier and resume the graph from that exact moment
earlier = history[len(history) // 2]
print("A checkpoint from earlier held", len(earlier.values["messages"]), "messages")
replay_config = {"configurable": {"thread_id": "conversation-1",
"checkpoint_id": earlier.config["configurable"]["checkpoint_id"]}}
resumed = graph.invoke(None, replay_config)
print("Resumed from the past; the graph now holds", len(resumed["messages"]), "messages")
again = graph.invoke({"messages": [{"role": "user", "content": "What do you know about me?"}]}, replay_config)
print(again["messages"][-1].content)
print(again["spanish"])Adding a UI
Let's add a simple Gradio UI to bring this to life. The chat function kicks off one run of the graph per message, using a fixed thread_id so the conversation is remembered, and it shows the Spanish translation in italics beneath each reply.
def chat(user_input: str, history):
config = {"configurable": {"thread_id": "gradio-session4"}}
result = graph.invoke({"messages": [{"role": "user", "content": user_input}]}, config)
return f"{result['messages'][-1].content}\n\n*{result['spanish']}*"
gr.ChatInterface(chat).launch()Recap, and where we are heading
You have built graphs by hand: state with a reducer, nodes, edges, a conditional edge, a tool loop, a second LLM node, observability, memory in two flavors, and a way to inspect and replay the whole thing. This is the machinery that powers everything above it.
Tomorrow we reach Layer 3, our next step along the control to convenience spine. We will ask create_agent to build a graph exactly like the tool loop above, in a single line, and then we will render that graph and recognise it as our own.