Street Learner
AuthorLast Updated: a year ago
In the previous two parts, we built a strong foundation of LangGraph fundamentals—nodes, edges, message states, conditional routing, reducers, summarization loops, and graph orchestration.
Now we enter an advanced phase:
We integrate checkpointing and persistence so your LangGraph pipeline becomes:
This shift moves LangGraph from a learning tool to a production-ready workflow engine.
We will cover four major concepts:
A checkpoint is a saved execution state.
When an LLM graph runs, each step produces:
Without checkpointing, this information disappears after execution.
Checkpointing allows:
In real applications—customer chatbots, research agents, retrieval pipelines—you must preserve state between runs.
LangGraph supports multiple checkpointing backends:
We will build both today.
Threads allow multiple isolated executions to run through the same graph logic.
Why important?
Imagine a SaaS AI where:
Threads allow separation using:
config = {"configurable": {"thread_id": "unique_id"}}
LangGraph handles routing each thread:
This is production architecture for:
When you do not need database persistence and want fast runtime memory, use:
from langgraph.checkpoint.memory import InMemorySaver
Why used?
Short-term memory stores:
BUT:
That is okay for early builds.
Once execution runs with checkpointing enabled, LangGraph stores a snapshot for each step.
Snapshots contain:
Why valuable?
This becomes essential in real world application auditing.
SQLite persistence is one of LangGraph’s most powerful features.
It allows your graph to store state across:
SQLiteSaver creates a structured table holding:
It transforms LangGraph workflows from demo code into real AI applications.
Below is the complete architecture, built with your provided code and then explained in depth.
We:
import os
import sqlite3
from dotenv import load_dotenv
print(f"{'='*30}\nSECTION 0: Setup & Graph Definitions\n{'='*30}")
load_dotenv()
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
We now load the model:
chat = ChatOpenAI(
model="gpt-4o",
seed=365,
temperature=0,
max_completion_tokens=100
)
Now define a shared application State:
class State(MessagesState):
summary: str
Meaning:
We now define the nodes.
This node feeds a new human question into the conversation:
def ask_question(state: State) -> State:
print(f"\n-------> ENTERING ask_question:")
question = "What is your question?"
print(question)
if not state.get("summary"):
user_input = "Tell me about the history of the internet."
else:
user_input = "That's cool. Who invented the web?"
print(f"(Simulated Input): {user_input}")
return {"messages": [HumanMessage(user_input)]}
This node generates the response:
def chatbot(state: State) -> State:
print(f"\n-------> ENTERING chatbot:")
summary = state.get("summary", "")
system_message = f'''
Here's a quick summary of what's been discussed so far:
{summary}
Keep this in mind as you answer the next question.
'''
messages = [SystemMessage(system_message)] + state["messages"]
response = chat.invoke(messages)
response.pretty_print()
return {"messages": [response]}
This node builds conversation compression:
def summarize_messages(state: State) -> State:
print(f"\n-------> ENTERING summarize_messages:")
new_conversation = ""
for i in state["messages"]:
new_conversation += f"{i.type}: {i.content}\n\n"
summary_instructions = f'''
Update the ongoing summary by incorporating the new lines of conversation below.
Build upon the previous summary rather than repeating it,
so that the result reflects the most recent context and developments.
Respond only with the summary.
Previous Summary:
{state.get("summary", "")}
New Conversation:
{new_conversation}
'''
summary = chat.invoke([HumanMessage(summary_instructions)])
print(f"--- Updated Summary: {summary.content[:50]}... ---")
remove_messages = [RemoveMessage(id=i.id) for i in state["messages"]]
return {"messages": remove_messages, "summary": summary.content}
This ensures:
This is exactly how production chatbots behave.
Same graph reused across all persistence backends:
def build_graph():
graph = StateGraph(State)
graph.add_node("ask_question", ask_question)
graph.add_node("chatbot", chatbot)
graph.add_node("summarize_messages", summarize_messages)
graph.add_edge(START, "ask_question")
graph.add_edge("ask_question", "chatbot")
graph.add_edge("chatbot", "summarize_messages")
graph.add_edge("summarize_messages", END)
return graph
print(f"\n{'='*30}\nSECTION 1: InMemorySaver (Short-Term)\n{'='*30}")
memory_checkpointer = InMemorySaver()
graph_memory = build_graph().compile(checkpointer=memory_checkpointer)
config1 = {"configurable": {"thread_id": "1"}}
config2 = {"configurable": {"thread_id": "2"}}
print("--- Thread 1 Execution ---")
graph_memory.invoke(State(messages=[], summary=""), config1)
print("\n--- Thread 2 Execution (Independent) ---")
graph_memory.invoke(State(messages=[], summary=""), config2)
Results:
print(f"\n{'='*30}\nSECTION 2: State Snapshots (Inspecting History)\n{'='*30}")
graph_states = [i for i in graph_memory.get_state_history(config1)]
print(f"Number of snapshots found: {len(graph_states)}")
Each snapshot contains:
This is debugging gold.
print(f"\n{'='*30}\nSECTION 3: Long-Term Persistence (SQLite)\n{'='*30}")
db_path = "langgraph_memory.db"
con = sqlite3.connect(database=db_path, check_same_thread=False)
sqlite_checkpointer = SqliteSaver(con)
graph_sqlite = build_graph().compile(checkpointer=sqlite_checkpointer)
config_sqlite = {"configurable": {"thread_id": "persistence_demo_1"}}
graph_sqlite.invoke(State(messages=[], summary=""), config_sqlite)
This finally creates:
Your .db file now contains:
This script:
from langgraph.graph import START, END, StateGraph, MessagesState
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import sqlite3
# PATH
con = sqlite3.connect("memory_persist_demo.db", check_same_thread=False)
# LLM
llm = ChatOpenAI(model="gpt-4o")
class State(MessagesState):
summary: str
def ask(state: State):
return {"messages": [HumanMessage("Explain AI agents")]}
def answer(state: State):
messages = state["messages"]
res = llm.invoke(messages)
return {"messages": [res]}
def summarize(state: State):
summary = "\n".join([m.content for m in state["messages"]])
return {"summary": summary}
graph = StateGraph(State)
graph.add_node("ask", ask)
graph.add_node("answer", answer)
graph.add_node("summarize", summarize)
graph.add_edge(START, "ask")
graph.add_edge("ask", "answer")
graph.add_edge("answer", "summarize")
graph.add_edge("summarize", END)
compiled = graph.compile(checkpointer=SqliteSaver(con))
config = {"configurable": {"thread_id": "A123"}}
out = compiled.invoke(State(messages=[], summary=""), config)
print("SUMMARY SAVED:", out["summary"])
Run this file multiple times.
You will notice:
This is real AI persistence.
With this Part 3 chapter, you now know how to build enterprise-friendly LangGraph pipelines:
This is production engineering for:
In the previous two parts, we built a strong foundation of LangGraph fundamentals—nodes, edges, message states, conditional routing, reducers, summarization loops, and graph orchestration.
Street Learner
05 Feb 2025
In Part-1 of this LangGraph Blog Series, we understood the foundation of LangGraph — Graph structure, Nodes, Edges, Conditional Routing, State system, and Graph Execution. Now in Part-2, we upgrade our knowledge and turn LangGraph into a real conversation system.
Street Learner
02 Feb 2025
Modern AI workflows need more than just a prompt and a model call. Real applications require memory, state transitions, branching logic, routing decisions, and orchestration of multiple AI models. This is where LangGraph enters the scene.
Street Learner
30 Jan 2025