LangGraph

The Chain Limitation

Early LLM orchestration frameworks modelled pipelines as directed acyclic sequences of operations. This works for simple prompt-response tasks but breaks down as soon as an agent must loop, retry, or branch based on intermediate results.

• No native support for cycles or loops
• State must be passed manually between steps
• No built-in pause / resume semantics
• Difficult to insert human feedback mid-run
• Debugging long runs requires custom instrumentation

What Real Agents Need

A LangGraph application is a compiled StateGraph — a finite state machine whose transitions are driven by LLM outputs and tool results rather than hardcoded logic.

Three primitives compose every graph:

Schema Definition

Every graph is parameterised by a State schema. All nodes and edges receive an instance of this schema as their sole argument.

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

class AgentState(TypedDict):
    # add_messages is a reducer — it appends
    # rather than overwriting the list
    messages: Annotated[list, add_messages]
    tool_output: str
    iteration:  int

Reducers

When a node returns a partial dict, LangGraph merges each key using its reducer function. The default reducer is last-write-wins. Custom reducers enable append semantics, set union, or conflict resolution.

Update Semantics

A node only needs to return the keys it modifies — not the entire state object. This avoids race conditions in parallel branches and keeps node logic cohesive.

def call_llm(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    # return only the keys this node touches
    return {
        "messages": [response],  # reducer appends
        "iteration": state["iteration"] + 1,
    }

A node is any callable that accepts a state dict and returns a (partial) state dict. This uniformity means nodes are trivially composable and independently testable.

Built-in Node Types

Async & Parallel Nodes

Nodes support async def natively. When multiple nodes have no data dependency in the same super-step they execute in parallel within that super-step.

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)

# register functions as nodes
graph.add_node("reason", call_llm)
graph.add_node("act",    ToolNode(tools))

# direct edge: reason always proceeds to act
graph.add_edge(START,    "reason")
graph.add_edge("act",    "reason")   # ← cycle!

# conditional edge: route after reasoning
graph.add_conditional_edges(
    "reason",
    should_continue,   # router fn → str
    {"continue": "act", "end": END}
)

app = graph.compile()

Edge Types

Pregel Super-Steps

Execution proceeds in discrete super-steps, inspired by Google's Pregel BSP model. A super-step ends when all currently active nodes have completed. Nodes activated in the same super-step run concurrently.

Before a graph can be executed it must be compiled via graph.compile(). Compilation is lightweight but serves several critical roles.

Validation Checks

• All declared nodes are reachable from START
• No orphaned nodes (nodes with no outgoing or incoming edges)
• All conditional edge targets exist as registered nodes
• State schema is a valid TypedDict or Pydantic model

Runtime Configuration

Compile-time parameters attach persistent infrastructure to the immutable graph:

app = graph.compile(
    checkpointer = PostgresSaver(conn),
    interrupt_before = ["human_review"],
    interrupt_after  = ["tool_call"],
)

Output: CompiledGraph

The compiled object is a LangChain Runnable, giving it a standard interface regardless of internal complexity:

The compiled graph is immutable — runtime modifications that could destabilise execution are prevented by design.

LangGraph's checkpointer saves a complete snapshot of graph state after every super-step. Snapshots are keyed by (thread_id, checkpoint_id).

Checkpointer Backends

All backends implement a common BaseCheckpointSaver interface with .put, .get_tuple, and .list methods, making backends interchangeable with a single line change.

Capabilities Unlocked by Checkpointing

LangGraph provides two complementary mechanisms for injecting human oversight at arbitrary points in a running graph.

Interrupt / Breakpoints

Setting interrupt_before=["node_name"] or interrupt_after=["node_name"] at compile time causes execution to pause and serialise state before or after that node fires. The __interrupt__ key in the returned state carries a structured payload describing why execution paused.

# Resume after human provides approval
app.invoke(
    Command(resume={"approved": True}),
    config={"configurable": {"thread_id": tid}}
)

Dynamic interrupt() Call

Inside any node, interrupt(payload) pauses execution programmatically — no compile-time configuration required. The payload is surfaced to the caller.

State Editing

While paused, the caller can patch graph state via app.update_state() before resuming — enabling human correction of LLM-generated content mid-run.

LangGraph is designed from the ground up for token-by-token streaming — essential for responsive UIs in agentic applications.

Stream Modes

Multiple modes can be combined in a single call. Subgraph events are exposed by passing subgraphs=True.

astream_events

The highest-resolution API, emitting structured lifecycle events for every LLM call, tool invocation, and node transition. Powers real-time "thinking" indicators and step-by-step progress UIs.

async for event in app.astream_events(
    {"messages": [HumanMessage(query)]},
    config={"configurable": {"thread_id": "1"}},
    version="v2"
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        token = event["data"]["chunk"].content
        print(token, end="", flush=True)
    elif kind == "on_tool_start":
        print(f"\n[tool] {event['name']}")

While LangGraph can be deployed as a plain Python library, LangChain provides a managed LangGraph Platform (LangSmith Deployment) that wraps any compiled graph behind a production API server.

Platform Capabilities

• Automatic checkpointing — no manual PostgresSaver wiring
• Autoscaling background task workers for long-running runs
• Built-in execution queue for fan-out workloads
• REST & WebSocket endpoints auto-generated from the graph
• LangGraph Studio — visual graph debugger & execution explorer
• LangSmith integration — traces, evals, and production monitoring

LangGraph Studio

A local desktop GUI that renders the compiled graph topology, lets developers step through execution node-by-node, inspect state at each checkpoint, and replay from any prior step — dramatically reducing iteration time for complex agents.

Self-Hosted Deployment

LangGraph Server can be self-hosted as a Docker container using the langgraph-cli tool. It provides the same REST API and handles persistence, streaming, and task queuing independently.

# build and serve
langgraph build -t my-agent:latest
langgraph up --port 8000

Strengths

Trade-Offs

A complete, minimal ReAct agent with persistent memory and human approval before each tool call.

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver

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

# ── Nodes ──────────────────────────────────────
llm_with_tools = llm.bind_tools(tools)

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

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

# ── Graph construction ─────────────────────────
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "agent")
builder.add_conditional_edges(
    "agent", should_continue,
    {"tools": "tools", END: END}
)
builder.add_edge("tools", "agent")

# ── Compile with persistence + HITL ────────────
memory = InMemorySaver()
app = builder.compile(
    checkpointer      = memory,
    interrupt_before  = ["tools"],   # pause for approval
)

# ── Invoke ─────────────────────────────────────
cfg = {"configurable": {"thread_id": "t1"}}
state = app.invoke(
    {"messages": [HumanMessage("Research LangGraph")]},
    config=cfg
)
# state["__interrupt__"] signals pending approval
# after human approval:
app.invoke(Command(resume=True), config=cfg)

This pattern — under 40 lines — captures the full LangGraph capability surface: typed state, reducers, conditional edges, cyclic execution, checkpointing, and human-in-the-loop interrupts.

LangGraph ships a first-class TypeScript / JavaScript SDK (@langchain/langgraph) with API parity to the Python version. The same graph primitives — state, nodes, edges, checkpointers — work identically.

Installation

npm install @langchain/langgraph @langchain/core
# or
yarn add @langchain/langgraph @langchain/core

State Definition (Annotation API)

TypeScript uses an Annotation builder instead of Python's TypedDict. Reducers are declared inline.

import { Annotation } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";

const AgentState = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: (curr, update) => [...curr, ...update],
    default: () => [],
  }),
  iteration: Annotation<number>({
    reducer: (_curr, update) => update,
    default: () => 0,
  }),
});

Graph Construction

import { StateGraph, START, END } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";

const graph = new StateGraph(AgentState)
  .addNode("reason", callLlm)
  .addNode("act", new ToolNode(tools))
  .addEdge(START, "reason")
  .addEdge("act", "reason")
  .addConditionalEdges("reason", shouldContinue, {
    continue: "act",
    end: END,
  });

const app = graph.compile();

Node Implementation

async function callLlm(
  state: typeof AgentState.State
) {
  const response = await llmWithTools.invoke(
    state.messages
  );
  return {
    messages: [response],
    iteration: state.iteration + 1,
  };
}

function shouldContinue(
  state: typeof AgentState.State
): string {
  const last = state.messages[state.messages.length - 1];
  return last.tool_calls?.length ? "continue" : "end";
}

Streaming (Token-Level)

const stream = await app.stream(
  { messages: [new HumanMessage("Research AI")] },
  { configurable: { thread_id: "t1" },
    streamMode: "messages" }
);

for await (const [message, metadata] of stream) {
  if (message.content) {
    process.stdout.write(String(message.content));
  }
}

Checkpointing

import { MemorySaver } from "@langchain/langgraph";

const memory = new MemorySaver();
const app = graph.compile({
  checkpointer: memory,
  interruptBefore: ["act"],
});

// Resume after human approval
await app.invoke(
  { messages: [] },   // empty — resumes from checkpoint
  { configurable: { thread_id: "t1" } }
);

Full ReAct Agent (TypeScript)

import { ChatOpenAI } from "@langchain/openai";
import { createReactAgent } from
  "@langchain/langgraph/prebuilt";
import { TavilySearchResults } from
  "@langchain/community/tools/tavily_search";

const llm = new ChatOpenAI({ model: "gpt-4o" });
const tools = [new TavilySearchResults()];

const agent = createReactAgent({
  llm,
  tools,
});

const result = await agent.invoke({
  messages: [
    { role: "user", content: "What is LangGraph?" }
  ],
});

Deployment

Custom Reducer (Python)

from typing import Annotated

def merge_dicts(current: dict, update: dict) -> dict:
    """Deep merge — update wins on conflict."""
    return {**current, **update}

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    findings: Annotated[dict, merge_dicts]
    sources:  Annotated[list, lambda a, b: a + b]

Scatter-Gather (Python)

from langgraph.types import Send

def fan_out(state: State) -> list[Send]:
    """Dispatch one worker per query."""
    return [
        Send("search_worker", {"query": q})
        for q in state["queries"]
    ]

graph.add_conditional_edges(
    "planner", fan_out
)
graph.add_edge("search_worker", "aggregator")

Supervisor Multi-Agent (Python)

def supervisor(state: State) -> dict:
    """Central LLM decides which agent to call."""
    response = llm.invoke([
        SystemMessage("Route to: researcher, "
                      "coder, or FINISH"),
        *state["messages"]
    ])
    return {"next_agent": response.content}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", research_agent)
graph.add_node("coder", coding_agent)

graph.add_conditional_edges("supervisor",
    lambda s: s["next_agent"],
    {"researcher": "researcher",
     "coder": "coder",
     "FINISH": END})
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")

Dynamic interrupt() (Python)

from langgraph.types import interrupt

def risky_action(state: State) -> dict:
    approval = interrupt({
        "action": "delete_file",
        "path": state["target_path"],
    })
    if not approval.get("approved"):
        return {"messages": [AIMessage("Aborted.")]}
    # proceed with action...
    return {"messages": [AIMessage("Done.")]}

Key Design Decisions

• Low-level primitives — no opinionated cognitive architecture imposed
• Immutable compiled graph — behaviour is consistent across all executions
• Runnable interface — composable with LangChain LCEL pipelines
• Pregel super-steps — clean parallelism model without explicit thread management
• Single checkpointer at root — automatically propagates to all subgraphs

References

LangGraph Documentation — docs.langchain.com
LangGraph GitHub — github.com/langchain-ai/langgraph
Malewicz et al., "Pregel: A System for Large-Scale Graph Processing," SIGMOD 2010
LangChain Academy — academy.langchain.com
LangGraph Platform — langchain.com/langgraph
LangSmith Deployment Guide — docs.smith.langchain.com