LangGraph Cheat Sheet — Stateful Agents & Graphs — LangGraph

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
import os
from langchain_openai import ChatOpenAI

class AgentState(TypedDict):
    messages: list
    counter: int

def node_a(state: AgentState) -> AgentState:
    """First processing step"""
    return {**state, "counter": state["counter"] + 1}

def node_b(state: AgentState) -> AgentState:
    """Second processing step"""
    return {**state, "counter": state["counter"] * 2}

graph = StateGraph(AgentState)
graph.add_node("step_a", node_a)
graph.add_node("step_b", node_b)

graph.add_edge(START, "step_a")
graph.add_edge("step_a", "step_b")
graph.add_edge("step_b", END)

compiled = graph.compile()
result = compiled.invoke({"messages": [], "counter": 0})
print(result["counter"])  # 2

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

class AgentState(TypedDict):
    messages: list
    sentiment: str

def analyze_sentiment(state: AgentState) -> AgentState:
    text = state["messages"][-1]["content"]
    sentiment = "positive" if "good" in text.lower() else "negative"
    return {**state, "sentiment": sentiment}

def route_based_sentiment(state: AgentState) -> Literal["happy_path", "escalate"]:
    """Conditional router—returns next node name"""
    return "happy_path" if state["sentiment"] == "positive" else "escalate"

def happy_path_node(state: AgentState) -> AgentState:
    state["messages"].append({"role": "assistant", "content": "Glad to help!"})
    return state

def escalate_node(state: AgentState) -> AgentState:
    state["messages"].append({"role": "assistant", "content": "Let me escalate this."})
    return state

graph = StateGraph(AgentState)
graph.add_node("analyze", analyze_sentiment)
graph.add_node("happy_path", happy_path_node)
graph.add_node("escalate", escalate_node)

graph.add_edge(START, "analyze")
graph.add_conditional_edges("analyze", route_based_sentiment)
graph.add_edge("happy_path", END)
graph.add_edge("escalate", END)

compiled = graph.compile()
result = compiled.invoke({"messages": [{"role": "user", "content": "This is good"}], "sentiment": ""})
print(result["messages"][-1]["content"])  # Glad to help!

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

class AgentState(TypedDict):
    messages: list
    iterations: int
    stop: bool

def should_continue(state: AgentState) -> Literal["tools", END]:
    """Break loop if stop=True or max iterations reached"""
    if state["stop"] or state["iterations"] >= 3:
        return END
    return "tools"

def agent_node(state: AgentState) -> AgentState:
    """Simulate agent reasoning step"""
    state["iterations"] += 1
    state["messages"].append({"role": "assistant", "content": f"Step {state['iterations']}"})
    if state["iterations"] >= 2:
        state["stop"] = True
    return state

def tool_node(state: AgentState) -> AgentState:
    """Execute tools"""
    state["messages"].append({"role": "tool", "content": "Tool result"})
    return state

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

compiled = graph.compile()
result = compiled.invoke({"messages": [], "iterations": 0, "stop": False})
print(f"Completed in {result['iterations']} iterations")

from langgraph.graph.message_graph import MessageGraph
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(model="gpt-4o", api_key=os.environ["OPENAI_API_KEY"])

# MessageGraph automatically manages messages as state
graph = MessageGraph()
graph.add_node("agent", llm)
graph.add_edge("agent", "__end__")
graph.set_entry_point("agent")

compiled = graph.compile()

# Invoke with just message content
result = compiled.invoke({"messages": [{"role": "user", "content": "What is 2+2?"}]})
print(result["messages"][-1].content)

from langgraph.graph import StateGraph, START, END
from langgraph.types import Interrupt
from typing import TypedDict

class AgentState(TypedDict):
    messages: list
    approved: bool

def propose_action(state: AgentState) -> AgentState:
    state["messages"].append({"role": "assistant", "content": "Proposed action"})
    # This raises an Interrupt—execution pauses here
    raise Interrupt({"status": "awaiting_approval"})

def execute_action(state: AgentState) -> AgentState:
    if not state["approved"]:
        state["messages"].append({"role": "system", "content": "Action rejected"})
        return state
    state["messages"].append({"role": "system", "content": "Action executed"})
    return state

graph = StateGraph(AgentState)
graph.add_node("propose", propose_action)
graph.add_node("execute", execute_action)

graph.add_edge(START, "propose")
graph.add_edge("propose", "execute")
graph.add_edge("execute", END)

compiled = graph.compile()

# First invocation pauses at interrupt
thread_id = "thread_1"
try:
    result = compiled.invoke({"messages": [], "approved": False}, {"configurable": {"thread_id": thread_id}})
except Exception as e:
    print(f"Paused: {e}")

# Resume with human approval
final = compiled.invoke({"messages": [], "approved": True}, {"configurable": {"thread_id": thread_id}})

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
import os

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

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

llm = ChatOpenAI(model="gpt-4o", api_key=os.environ["OPENAI_API_KEY"])
tools = [multiply, add]

# ReAct agent: LLM + tool loop built-in
agent = create_react_agent(llm, tools)

result = agent.invoke({"messages": [{"role": "user", "content": "What is (5 * 3) + 2?"}]})
print(result["messages"][-1].content)