LangGraph Cheat Sheet: State, Nodes, Branching, and Agent Control

This cheat sheet is a practical reference for designing LangGraph workflows that remain controlled, testable, and production-safe. LangGraph is most useful when the system is no longer a linear prompt pipeline and starts behaving like a stateful workflow engine with model reasoning inside it.

Installation

pip install langgraph langchain-openai langgraph-checkpoint-sqlite

1. Core Concepts

Concept	Type	Role
StateGraph	Graph container	Holds nodes, edges, state schema, compilation
TypedDict state	Shared memory	Typed contract between all nodes
Node	Python function state → state	One testable unit of work
add_edge	Unconditional transition	Always goes from A to B
add_conditional_edges	Branching	Routing function decides the next node
MemorySaver	In-memory checkpointer	State persistence across turns (dev)
SqliteSaver	SQLite checkpointer	State persistence (single-process prod)
interrupt_before	Human-in-the-loop	Pause before a node executes
Command	Node return type	Node that modifies graph flow explicitly

2. Minimal Graph

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

# 1. Define state schema
class State(TypedDict):
    question: str
    answer:   str

# 2. Define nodes — each receives state and returns updated state
llm = ChatOpenAI(model="gpt-4o-mini")

def generate(state: State) -> State:
    state["answer"] = llm.invoke(state["question"]).content
    return state

# 3. Build and compile graph
graph = StateGraph(State)
graph.add_node("generate", generate)
graph.set_entry_point("generate")   # first node to run
graph.add_edge("generate", END)      # exit condition

app = graph.compile()
result = app.invoke({"question": "What is LangGraph?", "answer": ""})
print(result["answer"])

3. State Design

from typing import Annotated
from langgraph.graph.message import add_messages
import operator

class AgentState(TypedDict):
    # Annotated fields use reducers instead of last-write-wins
    messages:      Annotated[list, add_messages]   # append-only
    retrieved_docs: Annotated[list, operator.add]  # extend list
    plan:          list[str]
    tool_results:  list[dict]
    current_step:  str | None
    retry_count:   int
    needs_approval: bool
    error:         str | None
    final_answer:  str | None

Reducer Reference

Reducer	Behavior	Use For
add_messages	Appends new messages, deduplicates by ID	Chat message history
operator.add	Extends list	Accumulating results
lambda a, b: {**a, **b}	Merges dicts	Metadata accumulation
None (default)	Last write wins	Simple scalar fields

4. Conditional Edges (Branching)

def router(state: AgentState) -> str:
    """Returns the name of the next node to execute."""
    if state.get("error"):
        return "error_handler"
    if state["retry_count"] >= 3:
        return "escalate"
    if state["needs_approval"]:
        return "human_review"
    if state["final_answer"]:
        return END
    return "reasoning"

graph.add_conditional_edges(
    "tool_node",    # source node
    router,         # routing function: state → str
    {               # optional: map return value → node name
        "error_handler": "error_handler",
        "escalate":       "escalate",
        "human_review":   "human_review",
        "reasoning":      "reasoning",
        END:              END,
    }
)

5. Persistence & Memory

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# In-memory (dev / tests)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# SQLite (production, single process)
with SqliteSaver.from_conn_string("./state.db") as saver:
    app = graph.compile(checkpointer=saver)

# thread_id ties all invocations to the same conversation
config = {"configurable": {"thread_id": "user-session-42"}}

app.invoke({"messages": [{"role": "user", "content": "Hello"}]}, config=config)
app.invoke({"messages": [{"role": "user", "content": "Follow up"}]}, config=config)
# State is automatically restored from checkpoint on the second call

# Inspect stored state
snapshot = app.get_state(config)
print(snapshot.values)          # current state dict
print(snapshot.next)            # next node(s) to run
print(snapshot.config)          # checkpoint metadata

# List all historical checkpoints
for checkpoint in app.get_state_history(config):
    print(checkpoint.created_at, checkpoint.values.get("current_step"))

6. Human-in-the-Loop

# Pause BEFORE specific nodes — execution suspends, state is saved
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["human_review", "execute_payment"],  # list of node names
    interrupt_after=["data_extraction"],                    # pause after instead
)

# Run until an interrupt
state = app.invoke(initial_state, config=config)
# → graph pauses, returns current state

# Inspect what the graph wants to do next
snapshot = app.get_state(config)
print("Next node:", snapshot.next)
print("State:",     snapshot.values)

# Optionally modify state before resuming
app.update_state(config, {"needs_approval": False, "retry_count": 0})

# Resume from checkpoint (None = use saved state)
final = app.invoke(None, config=config)

7. Streaming Modes

# Full state after every node
for state in app.stream(initial, stream_mode="values"):
    print(state)

# Only the delta (changed keys) after each node
for update in app.stream(initial, stream_mode="updates"):
    print(update)

# Token-level streaming from LLM nodes
for chunk, metadata in app.stream(initial, stream_mode="messages"):
    if metadata.get("langgraph_node") == "generate":
        print(chunk.content, end="", flush=True)

# All internal events — debugging
for event in app.stream(initial, stream_mode="debug"):
    print(event["type"], event.get("step"))

stream_mode	Returns	Best For
"values"	Full state dict after each node	Observing state evolution
"updates"	Delta dict after each node	Lightweight monitoring
"messages"	Token-level chunks + metadata	Streaming UI
"debug"	All internal events	Debugging graph execution

8. Subgraphs

from langgraph.graph import StateGraph

# Build a self-contained subgraph
sub_graph = StateGraph(SubState)
sub_graph.add_node("step_a", step_a)
sub_graph.add_node("step_b", step_b)
sub_graph.set_entry_point("step_a")
sub_graph.add_edge("step_a", "step_b")
sub_graph.add_edge("step_b", END)
sub_app = sub_graph.compile()

# Use the compiled subgraph as a node in the parent graph
def run_sub(parent_state: ParentState) -> ParentState:
    result = sub_app.invoke({"query": parent_state["question"]})
    parent_state["sub_result"] = result["answer"]
    return parent_state

parent = StateGraph(ParentState)
parent.add_node("sub_workflow", run_sub)

9. Prebuilt ReAct Agent

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

@tool
def search(query: str) -> str:
    """Search the web for up-to-date information."""
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression. e.g. '2 * (3 + 4)'"""
    return str(eval(expression))   # use a safe evaluator in production

llm   = ChatOpenAI(model="gpt-4o-mini")
agent = create_react_agent(
    llm,
    tools=[search, calculator],
    state_modifier="You are a helpful research assistant.",   # system prompt
    checkpointer=MemorySaver(),       # enable multi-turn memory
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "What is 15% of 2340?"}]
})
print(result["messages"][-1].content)

10. Error Handling Pattern

def safe_tool_node(state: AgentState) -> AgentState:
    try:
        result = call_external_api(state["params"])
        state["tool_results"].append({"status": "ok", "data": result})
        state["error"] = None
    except TimeoutError:
        state["error"]       = "tool_timeout"
        state["retry_count"] += 1
    except PermissionError:
        state["error"]          = "unauthorized"
        state["needs_approval"] = True
    except Exception as e:
        state["error"] = f"unexpected: {str(e)}"
    return state

def retry_router(state: AgentState) -> str:
    if state["error"] == "tool_timeout" and state["retry_count"] < 3:
        return "tool_node"       # retry
    if state["error"] == "unauthorized":
        return "human_review"    # escalate
    if state["error"]:
        return "error_node"      # terminal failure
    return "answer_node"         # success

graph.add_conditional_edges("tool_node", retry_router)

11. Full API Reference

Method	Signature	Effect
add_node	(name, fn)	Register a callable as a graph node
add_edge	(from, to)	Unconditional transition
add_conditional_edges	(from, fn, map?)	Routing transition
set_entry_point	(name)	Designate first node
set_finish_point	(name)	Mark a node as terminal
compile	(checkpointer=, interrupt_before=, interrupt_after=)	Build executable graph
invoke	(state, config)	Synchronous run to completion
stream	(state, config, stream_mode=)	Streaming run
aget_state	(config)	Inspect current checkpoint
update_state	(config, values, as_node=)	Modify paused state
get_state_history	(config, limit=)	List all past checkpoints

12. Production RAG Agent with LangGraph

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

# Initialize retriever
embeddings  = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
retriever   = vectorstore.as_retriever(search_kwargs={"k": 5})

class RAGAgentState(TypedDict):
    messages:        Annotated[list, add_messages]
    retrieved_docs:  list[str]
    generation_step: int
    needs_more_info: bool

@tool
def search_knowledge_base(query: str) -> str:
    """Search the knowledge base for relevant information about the query."""
    docs = retriever.invoke(query)
    return "\n\n".join(f"[Source {i+1}]: {d.page_content}" for i, d in enumerate(docs))

@tool
def web_search(query: str) -> str:
    """Search the web for current information not in the knowledge base."""
    # Integrate with Tavily or SerpAPI
    return f"Web results for: {query} (integrate with real search API)"

tools     = [search_knowledge_base, web_search]
llm       = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
tool_node = ToolNode(tools)

SYSTEM_PROMPT = """You are a helpful research assistant. 
Use the search_knowledge_base tool to find relevant information.
If the knowledge base doesn't have the answer, use web_search.
Always cite sources in your final answer."""

def agent_node(state: RAGAgentState) -> RAGAgentState:
    messages  = [{"role": "system", "content": SYSTEM_PROMPT}] + state["messages"]
    response  = llm.invoke(messages)
    return {
        "messages":        [response],
        "generation_step": state["generation_step"] + 1,
    }

def should_continue(state: RAGAgentState) -> str:
    last_msg = state["messages"][-1]
    if state["generation_step"] >= 6:   # safety limit
        return END
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return END

graph = StateGraph(RAGAgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app    = graph.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "rag-session-001"}}
result = app.invoke({
    "messages": [{"role": "user", "content": "What is our remote work policy?"}],
    "retrieved_docs": [],
    "generation_step": 0,
    "needs_more_info": False,
}, config=config)
print(result["messages"][-1].content)

---

13. Multi-Agent Supervisor Pattern

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

AGENTS = ["researcher", "writer", "reviewer"]

class SupervisorState(TypedDict):
    task:          str
    agent_outputs: dict[str, str]
    next_agent:    str | None
    iteration:     int
    final_answer:  str | None

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def supervisor_node(state: SupervisorState) -> SupervisorState:
    """Decides which agent to run next."""
    completed = list(state["agent_outputs"].keys())
    prompt = f"""
Task: {state["task"]}
Completed agents: {completed}
Available agents: {AGENTS}

Which agent should run next? 
- researcher: gathers facts and information
- writer: produces a draft based on research
- reviewer: reviews and improves the draft

If all agents have completed, return "FINISH".
Return only the agent name or "FINISH".
"""
    decision = llm.invoke(prompt).content.strip()
    return {"next_agent": None if decision == "FINISH" else decision}

def researcher_node(state: SupervisorState) -> SupervisorState:
    result = llm.invoke(f"Research this topic thoroughly: {state['task']}").content
    return {"agent_outputs": {**state["agent_outputs"], "researcher": result}}

def writer_node(state: SupervisorState) -> SupervisorState:
    research = state["agent_outputs"].get("researcher", "")
    result   = llm.invoke(f"Write a professional article based on:\n{research}\n\nTopic: {state['task']}").content
    return {"agent_outputs": {**state["agent_outputs"], "writer": result}}

def reviewer_node(state: SupervisorState) -> SupervisorState:
    draft  = state["agent_outputs"].get("writer", "")
    result = llm.invoke(f"Review and improve this draft:\n{draft}").content
    return {
        "agent_outputs": {**state["agent_outputs"], "reviewer": result},
        "final_answer":  result,
    }

def route_supervisor(state: SupervisorState) -> str:
    if state["next_agent"] is None or state["iteration"] >= 6:
        return END
    return state["next_agent"]

graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer",     writer_node)
graph.add_node("reviewer",   reviewer_node)
graph.set_entry_point("supervisor")

for agent in AGENTS:
    graph.add_edge(agent, "supervisor")

graph.add_conditional_edges("supervisor", route_supervisor, {
    "researcher": "researcher",
    "writer":     "writer",
    "reviewer":   "reviewer",
    END:           END,
})

app = graph.compile()
result = app.invoke({
    "task":          "Explain quantum computing advances in 2024",
    "agent_outputs": {},
    "next_agent":    None,
    "iteration":     0,
    "final_answer":  None,
})
print(result["final_answer"])

---

14. Long-Running Workflows with Checkpointing

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict

class WorkflowState(TypedDict):
    job_id:          str
    documents:       list[str]
    processed_count: int
    summaries:       list[str]
    final_report:    str | None

def process_batch_node(state: WorkflowState) -> WorkflowState:
    """Process a batch of documents."""
    batch_size = 5
    start_idx  = state["processed_count"]
    batch      = state["documents"][start_idx : start_idx + batch_size]

    new_summaries = []
    for doc in batch:
        summary = llm.invoke(f"Summarize in 2 sentences: {doc[:500]}").content
        new_summaries.append(summary)

    return {
        "processed_count": state["processed_count"] + len(batch),
        "summaries":       state["summaries"] + new_summaries,
    }

def should_continue_processing(state: WorkflowState) -> str:
    if state["processed_count"] >= len(state["documents"]):
        return "compile_report"
    return "process_batch"

def compile_report_node(state: WorkflowState) -> WorkflowState:
    all_summaries = "\n\n".join(state["summaries"])
    report        = llm.invoke(f"Compile a comprehensive report from:\n{all_summaries}").content
    return {"final_report": report}

with SqliteSaver.from_conn_string("./jobs.db") as saver:
    graph = StateGraph(WorkflowState)
    graph.add_node("process_batch",  process_batch_node)
    graph.add_node("compile_report", compile_report_node)
    graph.set_entry_point("process_batch")
    graph.add_conditional_edges("process_batch", should_continue_processing)
    graph.add_edge("compile_report", END)

    app = graph.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "batch-job-001"}}

    # Start job
    state = app.invoke({
        "job_id":          "batch-001",
        "documents":       ["doc1...", "doc2...", "doc3..."] * 10,  # 30 docs
        "processed_count": 0,
        "summaries":       [],
        "final_report":    None,
    }, config=config)

    # Job interrupted? Resume from checkpoint:
    # state = app.invoke(None, config=config)  # picks up where it left off

---

15. Testing LangGraph Workflows

import pytest
from langgraph.checkpoint.memory import MemorySaver

def test_agent_handles_tool_failure():
    """Test that the agent gracefully handles a tool error."""

    @tool
    def failing_tool(query: str) -> str:
        """A tool that always fails."""
        raise RuntimeError("External service unavailable")

    memory = MemorySaver()
    app    = build_agent([failing_tool], memory=memory)
    config = {"configurable": {"thread_id": "test-001"}}

    result = app.invoke(
        {"messages": [{"role": "user", "content": "Use the tool"}], "retry_count": 0},
        config=config,
    )

    # The agent should either retry or gracefully degrade
    last_msg = result["messages"][-1].content
    assert "unavailable" in last_msg.lower() or "sorry" in last_msg.lower()
    assert result["retry_count"] <= 3

def test_human_approval_flow():
    """Test that human-in-the-loop correctly pauses and resumes."""
    memory = MemorySaver()
    app    = sensitive_app.compile(
        checkpointer=memory,
        interrupt_before=["execute_action"],
    )
    config = {"configurable": {"thread_id": "test-hitl-001"}}

    # First run — should stop before execute_action
    state = app.invoke({"request": "delete all Q1 data", "human_approved": None}, config)
    snapshot = app.get_state(config)
    assert snapshot.next == ("execute_action",)

    # Reject and resume
    app.update_state(config, {"human_approved": False})
    final = app.invoke(None, config)
    assert "rejected" in final["final_result"].lower()

---

16. Parallel Node Execution

from langgraph.graph import StateGraph, END
from typing import TypedDict
import asyncio

class ResearchState(TypedDict):
    query:       str
    web_result:  str | None
    kb_result:   str | None
    analysis:    str | None

async def web_search_node(state: ResearchState) -> ResearchState:
    """Async web search — runs in parallel with kb_search."""
    result = await async_web_search(state["query"])
    return {"web_result": result}

async def kb_search_node(state: ResearchState) -> ResearchState:
    """Knowledge base search — runs in parallel with web_search."""
    result = await async_kb_search(state["query"])
    return {"kb_result": result}

def synthesis_node(state: ResearchState) -> ResearchState:
    """Synthesize results from both search paths."""
    context = f"Web:\n{state['web_result']}\n\nKB:\n{state['kb_result']}"
    answer  = llm.invoke(f"Synthesize: {context}\n\nQuery: {state['query']}").content
    return {"analysis": answer}

# Build graph with parallel branches
graph = StateGraph(ResearchState)
graph.add_node("web_search",  web_search_node)
graph.add_node("kb_search",   kb_search_node)
graph.add_node("synthesize",  synthesis_node)

graph.set_entry_point("web_search")

graph.add_edge("web_search", "synthesize")
graph.add_edge("kb_search",  "synthesize")
graph.add_edge("synthesize", END)

# For true parallelism in LangGraph, use Send API:
from langgraph.constants import Send

def fan_out(state: ResearchState) -> list[Send]:
    return [
        Send("web_search", state),
        Send("kb_search",  state),
    ]

graph.add_conditional_edges("router", fan_out)

---

17. State Versioning and Migration

from langgraph.graph import StateGraph
from typing import TypedDict
from pydantic import BaseModel

# Version 1 state
class StateV1(TypedDict):
    messages:    list
    answer:      str | None

# Version 2 state (adds new field)
class StateV2(TypedDict):
    messages:      list
    answer:        str | None
    confidence:    float | None    # NEW in v2
    source_docs:   list            # NEW in v2

def migrate_v1_to_v2(old_state: dict) -> StateV2:
    """Migrate persisted v1 state to v2 schema."""
    return {
        "messages":    old_state.get("messages", []),
        "answer":      old_state.get("answer"),
        "confidence":  None,
        "source_docs": [],
    }

# Load old checkpoint and migrate
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("./state.db") as saver:
    app    = graph_v2.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "user-123"}}

    snapshot = app.get_state(config)
    if snapshot and "confidence" not in snapshot.values:
        migrated = migrate_v1_to_v2(snapshot.values)
        app.update_state(config, migrated)

---

18. Observability and Debugging

from langsmith import traceable
from langgraph.graph import StateGraph, END

# Method 1: LangSmith tracing (automatic with env vars set)
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"]    = "ls__your_key"

# All graph.compile().invoke() calls are automatically traced

# Method 2: Custom event logging
class LoggingGraph:
    def __init__(self, compiled_graph):
        self.graph = compiled_graph

    def invoke(self, state: dict, config: dict = None) -> dict:
        print(f"[GRAPH START] Initial state keys: {list(state.keys())}")
        for step in self.graph.stream(state, config, stream_mode="updates"):
            for node_name, update in step.items():
                print(f"[NODE: {node_name}] Updated keys: {list(update.keys())}")
        final = self.graph.invoke(state, config)
        print(f"[GRAPH END] Final state keys: {list(final.keys())}")
        return final

# Method 3: Replay a specific checkpoint
snapshot = app.get_state(config)
for step in app.get_state_history(config, limit=10):
    print(f"Step {step.config['configurable']['checkpoint_id'][:8]}: "
          f"next={step.next}, created={step.created_at}")

---

19. Deployment Patterns

FastAPI + LangGraph

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver

app_api = FastAPI()
memory  = MemorySaver()
agent   = compiled_graph.compile(checkpointer=memory)

class ChatRequest(BaseModel):
    session_id: str
    message:    str

class ChatResponse(BaseModel):
    answer:     str
    session_id: str

@app_api.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    config  = {"configurable": {"thread_id": request.session_id}}
    state   = {
        "messages": [{"role": "user", "content": request.message}],
        "retry_count": 0,
    }
    try:
        result = agent.invoke(state, config)
        answer = result["messages"][-1].content
        return ChatResponse(answer=answer, session_id=request.session_id)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app_api.get("/session/{session_id}")
async def get_session_state(session_id: str):
    config   = {"configurable": {"thread_id": session_id}}
    snapshot = agent.get_state(config)
    if not snapshot:
        raise HTTPException(status_code=404, detail="Session not found")
    return {
        "session_id": session_id,
        "next":       snapshot.next,
        "message_count": len(snapshot.values.get("messages", [])),
    }

---

20. Common Pitfalls

Pitfall	Symptom	Fix
Infinite loop	Graph never terminates	Add max_iterations guard in router
Missing state key	KeyError in node	Use .get() with defaults; validate state schema
Checkpoint grows unbounded	Memory leak in MemorySaver	Use SqliteSaver with retention policy
Race condition in parallel nodes	Inconsistent state	Use Annotated reducers; never mutate state in-place
Thread ID collision	Users share state	Use UUID-based thread IDs, never static strings
Node function not pure	Flaky behavior on retry	Ensure nodes are idempotent pure functions
interrupt_before ignored	No pause occurs	Must use checkpointer and call invoke(None, config) to resume

---

LangGraph vs LangChain Decision Guide

Scenario	Use	Reason
Simple prompt → response	LangChain LCEL	No state needed
RAG with one retrieval	LangChain	Linear pipeline
Multi-turn chat	LangChain + memory	Simple history
Retry logic, branching	LangGraph	Conditional edges
Human approval required	LangGraph	interrupt_before
Long-running job resumption	LangGraph + SqliteSaver	Checkpoint persistence
Multi-agent collaboration	LangGraph	State shared across agents
Complex agentic loops	LangGraph	Explicit cycle control

---

Quick State Design Reference

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator

# Minimal chat state
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]  # append-only

# Full agent state
class AgentState(TypedDict):
    messages:     Annotated[list, add_messages]
    docs:         Annotated[list, operator.add]  # extend
    plan:         list[str]                      # overwrite
    step:         int                            # overwrite
    error:        str | None                     # overwrite
    approved:     bool | None                    # overwrite
    final_answer: str | None                     # overwrite

# Rule: use Annotated reducers for accumulating data,
# bare types for last-write-wins scalar fields

Graph Pattern Quick Reference

Pattern	Implementation
Linear A→B→C	add_edge(A,B); add_edge(B,C)
Branch	add_conditional_edges(A, router_fn)
Retry loop	conditional_edges routing back to A
Fan-out	Send API in conditional_edges
Human pause	interrupt_before=["node_name"]
Subgraph	compiled_subgraph as node function
Persistent memory	SqliteSaver checkpointer

---

LangGraph Performance Patterns

# Pattern 1: Async node execution for I/O-bound operations
from langgraph.graph import StateGraph, END
import asyncio

async def async_retrieval_node(state):
    """Use async for retrieval to avoid blocking."""
    docs = await retriever.ainvoke(state["question"])
    return {"retrieved_docs": [d.page_content for d in docs]}

async def async_llm_node(state):
    """Async LLM call for non-blocking generation."""
    context  = "\n".join(state["retrieved_docs"])
    response = await llm.ainvoke(f"Context: {context}\nQuestion: {state['question']}")
    return {"answer": response.content}

# Pattern 2: State compression to avoid memory growth
def compress_state_node(state):
    """Trim state to prevent memory bloat in long conversations."""
    messages = state.get("messages", [])
    if len(messages) > 20:
        compressed = messages[:1] + messages[-10:]
        return {"messages": compressed}
    return {}

# Pattern 3: Batched graph execution
async def run_parallel_graphs(inputs: list[dict], compiled_graph) -> list[dict]:
    """Run the same graph on multiple inputs in parallel."""
    tasks = [
        compiled_graph.ainvoke(inp, config={"configurable": {"thread_id": str(i)}})
        for i, inp in enumerate(inputs)
    ]
    return await asyncio.gather(*tasks)

results = asyncio.run(run_parallel_graphs(
    [{"question": q, "retry_count": 0} for q in questions],
    app,
))

State Schema Best Practices

# DO: Use Annotated reducers for accumulating data
class GoodState(TypedDict):
    messages:  Annotated[list, add_messages]   # LangGraph-aware appender
    doc_ids:   Annotated[list, operator.add]   # simple list extension
    score:     float                            # scalar: last write wins
    error:     str | None                       # nullable scalar

# DON'T: Mutate state in-place inside a node
def bad_node(state):
    state["messages"].append(new_message)  # WRONG: mutates in-place
    return state                           # breaks checkpointing

# DO: Return only the changed fields
def good_node(state):
    return {"messages": [new_message]}    # CORRECT: LangGraph merges this

---

LangGraph Production Principles

1. State is the single source of truth: All inter-node communication flows through typed state
2. Nodes must be pure functions: Same input always produces same output; no hidden side effects
3. Use reducers for accumulation: Annotated[list, add_messages] prevents subtle bugs from in-place mutation
4. Name your thread IDs meaningfully: Use user session IDs, not random UUIDs, for debuggability
5. Always set max_iterations: An agent with no iteration cap can loop indefinitely and exhaust your budget
6. Persist in production: MemorySaver is for dev; use SqliteSaver or PostgresSaver for production
7. Test state machine logic separately: Unit-test routing functions independently from LLM calls
8. Use interrupt_before for auditability: Pause before destructive operations for human review

---

Key Takeaways

• State is the single source of truth; all inter-node communication flows through typed state
• Use Annotated reducers for accumulating data; plain types for scalar overwrite
• Never mutate state in-place inside nodes; always return the changed keys
• MemorySaver is for dev; use SqliteSaver or PostgresSaver in production
• interrupt_before requires a checkpointer to be effective
• Always set a max iteration/step limit to prevent infinite agentic loops
• LangGraph is the right choice when you need branching, cycles, persistence, or human-in-the-loop

---

Conclusion

LangGraph is the production-grade foundation for stateful, cyclic LLM applications. Where LangChain LCEL handles linear pipelines, LangGraph handles workflows that need branching, loops, persistence, human-in-the-loop approval, and multi-agent collaboration. The explicit state machine design — with typed states, named nodes, conditional edges, and checkpoint-based persistence — trades initial complexity for long-term observability and reliability. For any LLM system that needs to handle more than a single round-trip with the model, LangGraph’s discipline pays dividends.