This post was generated with the help of Claude (Anthropic), based on the LangChain Academy — Introduction to LangGraph course.

A practical reference from beginner to intermediate.

Part 1 — Beginner Link to heading

1. Chat Models & Messages Link to heading

Setup

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
  • temperature=0 → deterministic; closer to 1 → more creative
  • Swap ChatOpenAI for any other provider without changing downstream code

Message types

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
  • HumanMessage — user input
  • AIMessage — model response
  • SystemMessage — instructions for the model
  • ToolMessage — result returned from a tool call

Invocation

llm.invoke("hello")                          # string → auto-wrapped as HumanMessage
llm.invoke([HumanMessage(content="hello")])  # explicit message list
llm.stream([HumanMessage(content="hello")])  # stream chunks back

2. Tools & Tool Calling Link to heading

Define a tool — any Python function with a docstring

def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

llm_with_tools = llm.bind_tools([multiply])
  • The model reads the docstring + signature to understand when/how to call the tool
  • The model returns a tool call object; you must execute it yourself (or use ToolNode)

Search tool (Tavily)

from langchain_tavily import TavilySearch
tavily = TavilySearch(max_results=3)
results = tavily.invoke({"query": "What is LangGraph?"})

3. LangGraph Core Concepts Link to heading

LangGraph models workflows as a directed graph: nodes do work, edges control flow.

Key primitives

  • State — a shared dict (or dataclass/Pydantic model) passed through all nodes
  • Node — a Python function (state) -> partial_state_update
  • Edge — connects nodes; can be normal (always) or conditional (routing logic)
  • START / END — special nodes marking entry and exit

Minimal graph skeleton

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class State(TypedDict):
    value: str

def my_node(state: State):
    return {"value": state["value"] + " processed"}

builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)
graph = builder.compile()

graph.invoke({"value": "hello"})

4. Conditional Edges (Router Pattern) Link to heading

from typing import Literal

def route(state: State) -> Literal["node_a", "node_b"]:
    if state["value"] == "go_a":
        return "node_a"
    return "node_b"

builder.add_conditional_edges("entry_node", route)
  • Return value must be an exact node name (or END)
  • Use tools_condition from langgraph.prebuilt to route on tool calls

5. Using Messages as State Link to heading

MessagesState is LangGraph’s built-in state for chat — pre-wired with the add_messages reducer so new messages are appended, not overwritten.

from langgraph.graph import MessagesState
# equivalent to:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], add_messages]

Chain: LLM + tool execution

from langgraph.prebuilt import ToolNode, tools_condition

def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")   # loop back after tool execution
graph = builder.compile()

6. ReAct Agent Link to heading

The standard agent loop: Act → Observe → Reason → repeat

from langchain_core.messages import SystemMessage

sys_msg = SystemMessage(content="You are a helpful assistant.")

def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
  • The graph loops assistant → tools → assistant until the model stops calling tools
  • Add parallel_tool_calls=False to bind_tools() for sequential tool use

7. Short-term Memory (Checkpointer) Link to heading

Persist conversation across multiple .invoke() calls using a thread ID.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-123"}}
graph.invoke({"messages": [HumanMessage(content="Hi")]}, config)
graph.invoke({"messages": [HumanMessage(content="What did I just say?")]}, config)
  • Same thread_id → the graph resumes from the last checkpoint
  • Different thread_id → fresh conversation
  • For persistence across restarts use SqliteSaver or PostgresSaver

Part 2 — Intermediate Link to heading

8. State Schema Options Link to heading

Three ways to define state — pick based on your validation needs:

StyleSyntaxRuntime validation
TypedDictclass S(TypedDict): x: strNo
dataclass@dataclass class S: x: strNo
Pydanticclass S(BaseModel): x: strYes
  • Access fields: state["key"] for TypedDict/Pydantic, state.key for dataclass
  • Use Literal["a","b"] to constrain allowed values

9. Reducers Link to heading

Control how state is updated when a node returns a value.

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

class State(TypedDict):
    messages: Annotated[list, add_messages]  # append
    counter: int                              # overwrite (default)
  • Default: overwrite the key with the new value
  • add_messages: appends new messages, deduplicates by ID
  • Custom reducer: Annotated[list, lambda old, new: old + new]
  • Branching conflict: if two parallel nodes update the same key without a reducer, LangGraph raises an error

10. Multiple Schemas Link to heading

Use separate input/output schemas to hide internal state from callers.

from langgraph.graph import StateGraph

class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    answer: str

class PrivateState(InputState, OutputState):
    internal_scratchpad: str   # never exposed externally

builder = StateGraph(PrivateState, input=InputState, output=OutputState)
  • Nodes can still read/write internal_scratchpad
  • Callers only see question in, answer out

11. Trim & Filter Messages Link to heading

Prevent context window overflow in long conversations.

from langchain_core.messages import trim_messages, filter_messages

# Keep only the last N tokens
trimmed = trim_messages(state["messages"], max_tokens=100, strategy="last")

# Keep only HumanMessages
filtered = filter_messages(state["messages"], include_types=["human"])

# Remove a specific message by ID
from langchain_core.messages import RemoveMessage
return {"messages": [RemoveMessage(id=msg.id)]}

Summarization pattern — compress old messages instead of discarding them:

def summarize(state: State):
    summary_msg = f"Summarize this conversation: {state['messages']}"
    new_summary = llm.invoke(summary_msg).content
    # delete old messages, keep only the last 2 + new summary
    delete_msgs = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": new_summary, "messages": delete_msgs}

12. Streaming Link to heading

Three modes — use them together or separately:

# Stream node-level state updates
for chunk in graph.stream(inputs, config, stream_mode="updates"):
    print(chunk)

# Stream LLM tokens (requires RunnableConfig threading in <3.11)
for chunk in graph.stream(inputs, config, stream_mode="messages"):
    print(chunk[0].content, end="", flush=True)

# Stream both
for chunk in graph.stream(inputs, config, stream_mode=["updates","messages"]):
    ...

13. Human-in-the-Loop Link to heading

Static breakpoints — pause before/after specific nodes:

graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"]   # pause before tools node
)

# Resume after human review
graph.invoke(None, config)  # pass None to resume from checkpoint

Dynamic breakpoints — raise from inside a node:

from langgraph.errors import NodeInterrupt

def risky_node(state: State):
    if state["needs_approval"]:
        raise NodeInterrupt("Human approval required")
    ...

Edit state & inject feedback:

# Inspect current state
snapshot = graph.get_state(config)

# Overwrite a state key
graph.update_state(config, {"messages": [HumanMessage(content="Override")]})

# Inject a human message as if it came from a node
graph.update_state(config, {"messages": [HumanMessage(...)]}, as_node="human_node")

14. Time Travel Link to heading

Replay or branch from any prior checkpoint.

# List all checkpoints for a thread
history = list(graph.get_state_history(config))

# Get the state at a specific checkpoint
past_state = history[2]

# Replay from that checkpoint (creates a new branch)
graph.invoke(None, past_state.config)
  • Each checkpoint has a unique config you can pass directly to invoke
  • Useful for debugging: rewind to just before a bad decision and try again

15. Parallelization (Fan-out / Fan-in) Link to heading

Run nodes concurrently when they don’t depend on each other.

builder.add_edge("a", "b")   # b and c run in parallel
builder.add_edge("a", "c")
builder.add_edge("b", "d")   # d runs after both b and c finish
builder.add_edge("c", "d")
  • Requires a reducer on any state key that multiple parallel nodes write to, or you’ll get a conflict error
  • Results are merged at the fan-in point

16. Sub-graphs Link to heading

Encapsulate a full graph as a node in a parent graph; each has its own private state.

sub_builder = StateGraph(SubState)
# ... build sub_builder ...
sub_graph = sub_builder.compile()

parent_builder = StateGraph(ParentState)
parent_builder.add_node("sub", sub_graph)   # compiled graph as a node
  • Communication happens through overlapping state keys between parent and sub-graph
  • Sub-graph can read keys defined in its state that exist in the parent; it writes back via overlapping keys
  • Ideal for multi-agent systems where each agent team has its own internal state

17. Map-Reduce (Send API) Link to heading

Dynamically spawn parallel branches at runtime — useful when you don’t know the number of branches ahead of time.

from langgraph.types import Send

def generate_tasks(state: State):
    # Send creates a new branch for each item
    return [Send("process_item", {"item": x}) for x in state["items"]]

builder.add_conditional_edges("planner", generate_tasks)
  • Each Send("node", state) spawns an independent copy of node with its own state slice
  • Results accumulate in the parent via a reducer on the output key
  • Classic pattern: fan-out to workers, fan-in to aggregator

18. Long-term Memory (Memory Store) Link to heading

Persist facts across threads — independent of checkpointers.

from langgraph.store.memory import InMemoryStore
import uuid

store = InMemoryStore()

# Save
namespace = ("user_123", "memories")
store.put(namespace, str(uuid.uuid4()), {"fact": "user likes jazz"})

# Retrieve
memories = store.search(namespace)
print(memories[0].value)

# In a graph node: store is injected automatically
def chatbot(state: MessagesState, store: BaseStore):
    memories = store.search(("user", "facts"))
    context = "\n".join(m.value["fact"] for m in memories)
    ...
  • Compile with graph = builder.compile(checkpointer=..., store=store)
  • InMemoryStore for dev; swap for PostgresStore or custom for production

19. Memory Schemas Link to heading

Structure stored memories with a profile (single updatable object) or collection (list of items).

Profile — one record per user, continuously updated:

from pydantic import BaseModel

class UserProfile(BaseModel):
    name: str = ""
    food_preferences: list[str] = []

# Use Trustcall to update the profile without losing existing fields
from trustcall import create_extractor
extractor = create_extractor(llm, tools=[UserProfile], tool_choice="UserProfile")

Collection — append new items, each with its own UUID:

class MemoryItem(BaseModel):
    content: str
    category: str   # e.g. "todo", "fact", "preference"

# Store each item separately under the same namespace
store.put(namespace, str(uuid.uuid4()), item.dict())
  • Use profiles when you want a single evolving summary (e.g., user preferences)
  • Use collections when each memory is distinct (e.g., tasks, events, facts)

20. Deployment & LangGraph Platform Link to heading

Project structure required

my_app/
  graph.py          # your compiled graph
  langgraph.json    # config: which graphs to expose
  requirements.txt
  .env

langgraph.json example:

{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./graph.py:graph"
  }
}

CLI commands

pip install langgraph-cli
langgraph dev          # local dev server + Studio UI
langgraph build        # build Docker image
langgraph up           # start with docker-compose (redis + postgres + API)

Connect via SDK

from langgraph_sdk import get_client
client = get_client(url="http://localhost:8123")

thread = await client.threads.create()
run = await client.runs.create(thread["thread_id"], "my_agent", input={...})

21. Assistants (Configuration Variants) Link to heading

Assistants let you version agent configurations (system prompt, model, parameters) without changing code.

# In your graph, expose configurable fields
from langgraph.graph import RunnableConfig

def assistant_node(state, config: RunnableConfig):
    cfg = config.get("configurable", {})
    role = cfg.get("assistant_role", "default instructions")
    ...

# Create different assistants via SDK
assistant_a = await client.assistants.create(
    "my_agent",
    config={"configurable": {"assistant_role": "work tasks"}}
)
assistant_b = await client.assistants.create(
    "my_agent",
    config={"configurable": {"assistant_role": "personal tasks"}}
)

22. Double-Texting Strategies Link to heading

Handle concurrent runs on the same thread (user sends messages before prior run finishes).

StrategyBehavior
rejectReject the new run while the current run is active
enqueueQueue the new run; execute after current finishes
interruptCancel current run, start new one immediately
rollbackCancel and delete current run, start fresh
run = await client.runs.create(
    thread_id,
    "my_agent",
    input={...},
    multitask_strategy="enqueue"  # or "reject", "interrupt", "rollback"
)

Quick Reference Link to heading

ConceptImport
Chat modelfrom langchain_openai import ChatOpenAI
Messagesfrom langchain_core.messages import HumanMessage, AIMessage, SystemMessage
Graph builderfrom langgraph.graph import StateGraph, START, END
Built-in statefrom langgraph.graph import MessagesState
Add messages reducerfrom langgraph.graph.message import add_messages
Prebuilt agent nodesfrom langgraph.prebuilt import ToolNode, tools_condition
In-memory checkpointerfrom langgraph.checkpoint.memory import MemorySaver
SQLite checkpointerfrom langgraph.checkpoint.sqlite import SqliteSaver
Memory storefrom langgraph.store.memory import InMemoryStore
Dynamic interruptfrom langgraph.errors import NodeInterrupt
Map-reducefrom langgraph.types import Send
Search toolfrom langchain_tavily import TavilySearch