LangChain & LangGraph Basics

8 examples to get you started with LangGraph - 5 basic and 3 intermediate.

You will define state, wire two nodes, add a conditional edge, compile, invoke, and stream updates. Later examples touch reducers, checkpoints, and a tiny tool-style loop.

Prerequisites

Python 3.11+ recommended
Comfort with typed dicts and function callables
Optional LLM key only for intermediate example 8 (others stay model-free)

python -m venv .venv && source .venv/bin/activate
pip install -U langgraph
# Optional for LLM examples (verify package names at build):
# pip install -U langchain langchain-openai

Basic Examples

1. Define a Minimal State Schema

State is the shared object every node reads and updates.

from typing import TypedDict
 
class TicketState(TypedDict):
    text: str
    label: str
    response: str

Start with plain fields and no reducers.
Keep names domain-specific (label, not x).
Empty strings are fine defaults for demos; production may use None or enums.

Related: The LangGraph Mental Model

2. Write Two Nodes That Return Partial Updates

Nodes return dicts of changes, not a full copy of state (unless you choose to).

def classify(state: TicketState) -> dict:
    text = state["text"].lower()
    label = "billing" if "invoice" in text or "refund" in text else "general"
    return {"label": label}
 
def draft_reply(state: TicketState) -> dict:
    if state["label"] == "billing":
        body = "We received your billing request and will review it."
    else:
        body = "Thanks for writing in. A teammate will follow up."
    return {"response": body}

Each node owns one job.
Return only keys you intend to change.
Pure functions are easier to unit test without compiling the graph.

3. Connect START, Nodes, END, and Compile

Fixed edges define a linear path.

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("draft_reply", draft_reply)
builder.add_edge(START, "classify")
builder.add_edge("classify", "draft_reply")
builder.add_edge("draft_reply", END)
graph = builder.compile()
 
result = graph.invoke({
    "text": "Please refund invoice #4421",
    "label": "",
    "response": "",
})
print(result["label"], result["response"])

compile() builds the runnable runtime.
invoke runs to completion and returns final state.
Linear graphs are workflows; agents appear when routing becomes data-dependent.

4. Add One Conditional Edge

Route after classification without hardcoding a single next step inside the node.

def route_label(state: TicketState) -> str:
    return "billing_path" if state["label"] == "billing" else "general_path"
 
def billing_path(state: TicketState) -> dict:
    return {"response": "Billing queue: " + state["text"][:80]}
 
def general_path(state: TicketState) -> dict:
    return {"response": "General queue: " + state["text"][:80]}
 
builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("billing_path", billing_path)
builder.add_node("general_path", general_path)
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_label,
    {"billing_path": "billing_path", "general_path": "general_path"},
)
builder.add_edge("billing_path", END)
builder.add_edge("general_path", END)
graph = builder.compile()
 
print(graph.invoke({"text": "refund please", "label": "", "response": ""})["response"])

The router returns a key that maps to a node name.
Conditional edges make branches visible in diagrams and traces.
Keep router functions free of side effects.

Related: Branching and Conditional Edges

5. Stream Node Updates

Watch each superstep instead of only the final state.

for chunk in graph.stream(
    {"text": "Where is my invoice?", "label": "", "response": ""},
    stream_mode="updates",
    version="v2",  # LangGraph >= 1.1; omit or use v1 if needed (verify at build)
):
    # v2 chunks: {"type": "updates", "ns": (), "data": {...}}
    if isinstance(chunk, dict) and chunk.get("type") == "updates":
        print(chunk["data"])
    else:
        print(chunk)  # v1 raw updates dict

updates shows per-node partial writes.
values (not shown) streams full state after each step.
Prefer version="v2" for a uniform stream part shape when your pin supports it.

Intermediate Examples

6. Append Messages With a Reducer

Chat-style agents need append, not overwrite.

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
 
def greeter(state: ChatState) -> dict:
    return {"messages": [{"role": "assistant", "content": "Hello!"}]}
 
g = (
    StateGraph(ChatState)
    .add_node("greeter", greeter)
    .add_edge(START, "greeter")
    .add_edge("greeter", END)
    .compile()
)
out = g.invoke({"messages": [{"role": "user", "content": "hi"}]})
print(len(out["messages"]))  # user + assistant

add_messages merges message lists safely.
Passing only new messages from a node is the usual pattern.
Wrong reducer choice is a top source of "history disappeared" bugs.

7. Persist a Thread With InMemorySaver

Checkpoints let the same thread_id continue later.

from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "demo-1"}}
 
graph.invoke(
    {"text": "refund invoice", "label": "", "response": ""},
    config=config,
)
snapshot = graph.get_state(config)
print(snapshot.values.get("label"), snapshot.next)

InMemorySaver is for local demos; production needs SQLite/Postgres (or platform-managed persistence).
Always pass thread_id when a checkpointer is attached.
get_state is the first debugging tool after "what happened?"

Related: Persistence and Checkpoints

8. Sketch a Mini Agent Cycle (Model Optional)

A classic agent loop is a cycle: model node, tools node, conditional stop.

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
 
def call_model(state: AgentState) -> dict:
    # Swap for a real chat model + bind_tools in production (verify APIs at build)
    last = state["messages"][-1]["content"]
    if "2+2" in last.replace(" ", ""):
        return {"messages": [{
            "role": "assistant",
            "content": "",
            "tool_calls": [{
                "id": "1",
                "name": "add",
                "args": {"a": 2, "b": 2},
            }],
        }]}
    return {"messages": [{"role": "assistant", "content": "Done."}]}
 
def call_tools(state: AgentState) -> dict:
    # Demo tool execution only
    return {"messages": [{
        "role": "tool",
        "content": "4",
        "tool_call_id": "1",
        "name": "add",
    }]}
 
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    if last.get("tool_calls"):
        return "tools"
    return "__end__"
 
agent = StateGraph(AgentState)
agent.add_node("agent", call_model)
agent.add_node("tools", call_tools)
agent.add_edge(START, "agent")
agent.add_conditional_edges("agent", should_continue, {"tools": "tools", "__end__": END})
agent.add_edge("tools", "agent")
app = agent.compile()
print(app.invoke({"messages": [{"role": "user", "content": "What is 2+2?"}]})["messages"][-1])

Cycles are normal; ensure a path to END.
Real code uses LangChain (or other) tool-calling messages; structure stays the same.
Add max-iteration guards before production (timeouts, counters, or recursion limits).

Related: Building Your First Stateful Agent Graph

The LangGraph Mental Model - why state graphs
Building Your First Stateful Agent Graph - fuller recipe
Branching and Conditional Edges - multi-path routing
Observability with LangSmith - node-level traces
LangChain & LangGraph Best Practices - production habits

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

LangChain & LangGraph Basics

8 examples to get you started with LangGraph - 5 basic and 3 intermediate.

You will define state, wire two nodes, add a conditional edge, compile, invoke, and stream updates. Later examples touch reducers, checkpoints, and a tiny tool-style loop.

Prerequisites

Python 3.11+ recommended
Comfort with typed dicts and function callables
Optional LLM key only for intermediate example 8 (others stay model-free)

python -m venv .venv && source .venv/bin/activate
pip install -U langgraph
# Optional for LLM examples (verify package names at build):
# pip install -U langchain langchain-openai

Basic Examples

1. Define a Minimal State Schema

State is the shared object every node reads and updates.

from typing import TypedDict
 
class TicketState(TypedDict):
    text: str
    label: str
    response: str

Start with plain fields and no reducers.
Keep names domain-specific (label, not x).
Empty strings are fine defaults for demos; production may use None or enums.

Related: The LangGraph Mental Model

2. Write Two Nodes That Return Partial Updates

Nodes return dicts of changes, not a full copy of state (unless you choose to).

def classify(state: TicketState) -> dict:
    text = state["text"].lower()
    label = "billing" if "invoice" in text or "refund" in text else "general"
    return {"label": label}
 
def draft_reply(state: TicketState) -> dict:
    if state["label"] == "billing":
        body = "We received your billing request and will review it."
    else:
        body = "Thanks for writing in. A teammate will follow up."
    return {"response": body}

Each node owns one job.
Return only keys you intend to change.
Pure functions are easier to unit test without compiling the graph.

3. Connect START, Nodes, END, and Compile

Fixed edges define a linear path.

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("draft_reply", draft_reply)
builder.add_edge(START, "classify")
builder.add_edge("classify", "draft_reply")
builder.add_edge("draft_reply", END)
graph = builder.compile()
 
result = graph.invoke({
    "text": "Please refund invoice #4421",
    "label": "",
    "response": "",
})
print(result["label"], result["response"])

compile() builds the runnable runtime.
invoke runs to completion and returns final state.
Linear graphs are workflows; agents appear when routing becomes data-dependent.

4. Add One Conditional Edge

Route after classification without hardcoding a single next step inside the node.

def route_label(state: TicketState) -> str:
    return "billing_path" if state["label"] == "billing" else "general_path"
 
def billing_path(state: TicketState) -> dict:
    return {"response": "Billing queue: " + state["text"][:80]}
 
def general_path(state: TicketState) -> dict:
    return {"response": "General queue: " + state["text"][:80]}
 
builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("billing_path", billing_path)
builder.add_node("general_path", general_path)
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_label,
    {"billing_path": "billing_path", "general_path": "general_path"},
)
builder.add_edge("billing_path", END)
builder.add_edge("general_path", END)
graph = builder.compile()
 
print(graph.invoke({"text": "refund please", "label": "", "response": ""})["response"])

The router returns a key that maps to a node name.
Conditional edges make branches visible in diagrams and traces.
Keep router functions free of side effects.

Related: Branching and Conditional Edges

5. Stream Node Updates

Watch each superstep instead of only the final state.

for chunk in graph.stream(
    {"text": "Where is my invoice?", "label": "", "response": ""},
    stream_mode="updates",
    version="v2",  # LangGraph >= 1.1; omit or use v1 if needed (verify at build)
):
    # v2 chunks: {"type": "updates", "ns": (), "data": {...}}
    if isinstance(chunk, dict) and chunk.get("type") == "updates":
        print(chunk["data"])
    else:
        print(chunk)  # v1 raw updates dict

updates shows per-node partial writes.
values (not shown) streams full state after each step.
Prefer version="v2" for a uniform stream part shape when your pin supports it.

Intermediate Examples

6. Append Messages With a Reducer

Chat-style agents need append, not overwrite.

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
 
def greeter(state: ChatState) -> dict:
    return {"messages": [{"role": "assistant", "content": "Hello!"}]}
 
g = (
    StateGraph(ChatState)
    .add_node("greeter", greeter)
    .add_edge(START, "greeter")
    .add_edge("greeter", END)
    .compile()
)
out = g.invoke({"messages": [{"role": "user", "content": "hi"}]})
print(len(out["messages"]))  # user + assistant

add_messages merges message lists safely.
Passing only new messages from a node is the usual pattern.
Wrong reducer choice is a top source of "history disappeared" bugs.

7. Persist a Thread With InMemorySaver

Checkpoints let the same thread_id continue later.

from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "demo-1"}}
 
graph.invoke(
    {"text": "refund invoice", "label": "", "response": ""},
    config=config,
)
snapshot = graph.get_state(config)
print(snapshot.values.get("label"), snapshot.next)

InMemorySaver is for local demos; production needs SQLite/Postgres (or platform-managed persistence).
Always pass thread_id when a checkpointer is attached.
get_state is the first debugging tool after "what happened?"

Related: Persistence and Checkpoints

8. Sketch a Mini Agent Cycle (Model Optional)

A classic agent loop is a cycle: model node, tools node, conditional stop.

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
 
def call_model(state: AgentState) -> dict:
    # Swap for a real chat model + bind_tools in production (verify APIs at build)
    last = state["messages"][-1]["content"]
    if "2+2" in last.replace(" ", ""):
        return {"messages": [{
            "role": "assistant",
            "content": "",
            "tool_calls": [{
                "id": "1",
                "name": "add",
                "args": {"a": 2, "b": 2},
            }],
        }]}
    return {"messages": [{"role": "assistant", "content": "Done."}]}
 
def call_tools(state: AgentState) -> dict:
    # Demo tool execution only
    return {"messages": [{
        "role": "tool",
        "content": "4",
        "tool_call_id": "1",
        "name": "add",
    }]}
 
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    if last.get("tool_calls"):
        return "tools"
    return "__end__"
 
agent = StateGraph(AgentState)
agent.add_node("agent", call_model)
agent.add_node("tools", call_tools)
agent.add_edge(START, "agent")
agent.add_conditional_edges("agent", should_continue, {"tools": "tools", "__end__": END})
agent.add_edge("tools", "agent")
app = agent.compile()
print(app.invoke({"messages": [{"role": "user", "content": "What is 2+2?"}]})["messages"][-1])

Cycles are normal; ensure a path to END.
Real code uses LangChain (or other) tool-calling messages; structure stays the same.
Add max-iteration guards before production (timeouts, counters, or recursion limits).

Related: Building Your First Stateful Agent Graph

The LangGraph Mental Model - why state graphs
Building Your First Stateful Agent Graph - fuller recipe
Branching and Conditional Edges - multi-path routing
Observability with LangSmith - node-level traces
LangChain & LangGraph Best Practices - production habits

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

LangChain & LangGraph Basics

Buscar en la documentación

LangChain & LangGraph Basics