LangGraph Cheatsheet: StateGraph, Nodes and Checkpoints

# Core
pip install langgraph              # ≥ 0.2
pip install langchain-core         # message types

# Checkpointers — split into separate packages
pip install langgraph-checkpoint-sqlite     # SqliteSaver
pip install langgraph-checkpoint-postgres   # PostgresSaver

# Prebuilt agents
pip install langgraph-prebuilt     # create_react_agent

# env (whichever model you bind)
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-...

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

# State schema — Annotated[type, reducer] for merge semantics
class State(TypedDict):
    messages: Annotated[list, add_messages]    # append, dedupe by id
    counter:  Annotated[int,  add]              # numeric sum
    flags:    list[str]                         # default = replace

def step_one(state: State) -> dict:
    return {"counter": 1, "flags": ["seen"]}

def step_two(state: State) -> dict:
    return {"counter": state["counter"] + 1}

graph = StateGraph(State)
graph.add_node("one", step_one)
graph.add_node("two", step_two)
graph.add_edge(START, "one")
graph.add_edge("one", "two")
graph.add_edge("two", END)

app = graph.compile()
print(app.invoke({"messages": [], "counter": 0, "flags": []}))

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([add])

def call_model(state: MessagesState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

# Router: tool_use → run tools, else stop
def route(state: MessagesState) -> str:
    last = state["messages"][-1]
    return "tools" if last.tool_calls else END

graph = StateGraph(MessagesState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode([add]))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route, ["tools", END])
graph.add_edge("tools", "agent")

app = graph.compile()
out = app.invoke({"messages": [HumanMessage("What is 17 + 25?")]})
print(out["messages"][-1].content)

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

def draft(state: MessagesState) -> dict:
    return {"messages": [{"role": "assistant", "content": "Proposed reply: Yes."}]}

def human_review(state: MessagesState) -> dict:
    # Pauses the graph. Resume by passing a Command(resume=...) value.
    edited = interrupt({"proposal": state["messages"][-1].content})
    return {"messages": [{"role": "assistant", "content": edited}]}

graph = StateGraph(MessagesState)
graph.add_node("draft", draft)
graph.add_node("review", human_review)
graph.add_edge(START, "draft")
graph.add_edge("draft", "review")
graph.add_edge("review", END)

app = graph.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "t1"}}

# First run pauses at the interrupt
result = app.invoke({"messages": []}, cfg)
print(result["__interrupt__"])    # value the node interrupted with

# Resume with the human's edit
final = app.invoke(Command(resume="Approved with edits."), cfg)
print(final["messages"][-1].content)

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

class State(TypedDict):
    messages: Annotated[list, add_messages]

@tool
def search(q: str) -> str:
    """Stub web search."""
    return f"top result for '{q}': LangGraph is a state-machine library."

llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([search])

def agent(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

def needs_tool(state: State) -> str:
    return "tools" if state["messages"][-1].tool_calls else END

g = StateGraph(State)
g.add_node("agent", agent)
g.add_node("tools", ToolNode([search]))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", needs_tool, ["tools", END])
g.add_edge("tools", "agent")

with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    app = g.compile(checkpointer=saver)
    cfg = {"configurable": {"thread_id": "user_42"}}
    for chunk in app.stream(
        {"messages": [HumanMessage("What is LangGraph?")]}, cfg,
    ):
        print(chunk)