langgraph

amite's avatarfrom amite

LangGraph low-level agent orchestration. Build stateful, durable agents with graphs, persistence, streaming, and human-in-the-loop.

⭐0stars🔀0forks📁View on GitHub🕐Updated Dec 18, 2025⬇️1downloads

When & Why to Use This Skill

LangGraph is a high-performance, low-level orchestration framework designed for building stateful, multi-step AI agents. By modeling workflows as graphs with nodes and edges, it provides developers with granular control over agent behavior, enabling the creation of complex, durable applications that support persistence, streaming, and human-in-the-loop interactions.

Use Cases

  • Complex Multi-Agent Systems: Designing workflows where multiple specialized agents collaborate through shared state and conditional logic transitions.
  • Human-in-the-Loop Workflows: Implementing manual intervention points for sensitive operations, allowing humans to review, edit, or approve agent actions before execution.
  • Long-Running Stateful Agents: Building resilient applications that maintain context across multiple sessions using checkpointers for persistence and error recovery.
  • Advanced RAG Pipelines: Creating sophisticated retrieval-augmented generation loops that involve iterative refinement, self-correction, and dynamic routing based on document relevance.
namelanggraph
description"LangGraph low-level agent orchestration. Build stateful, durable agents with graphs, persistence, streaming, and human-in-the-loop."

LangGraph Development

Source: https://github.com/langchain-ai/docs (src/oss/langgraph/)

LangGraph is the low-level orchestration framework for building stateful, long-running agents. It models workflows as graphs with nodes (functions) and edges (transitions). Use LangGraph when you need fine-grained control over agent behavior; use LangChain's create_agent for simpler use cases.

Core Concepts

  • State: Shared data structure representing the current snapshot
  • Nodes: Functions that process state and return updates
  • Edges: Determine which node runs next (conditional or fixed)

"Nodes do the work, edges tell what to do next."

Quick Start

from langgraph.graph import StateGraph, MessagesState, START, END

def call_model(state: MessagesState):
    # Your LLM logic here
    return {"messages": [{"role": "ai", "content": "Hello!"}]}

graph = StateGraph(MessagesState)
graph.add_node("model", call_model)
graph.add_edge(START, "model")
graph.add_edge("model", END)

app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Hi!"}]})

State Definition

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph
from operator import add

class State(TypedDict):
    messages: Annotated[list, add]  # Reducer: append to list
    count: int                       # Overwrite on update

graph = StateGraph(State)

Using MessagesState (Convenience)

from langgraph.graph import StateGraph, MessagesState

# MessagesState has a pre-defined "messages" key with append reducer
graph = StateGraph(MessagesState)

Adding Nodes

def node_a(state: State) -> dict:
    """Nodes receive state and return updates."""
    return {"count": state["count"] + 1}

def node_b(state: State) -> dict:
    return {"messages": [{"role": "ai", "content": f"Count is {state['count']}"}]}

graph.add_node("a", node_a)
graph.add_node("b", node_b)

Adding Edges

Fixed Edges

from langgraph.graph import START, END

graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", END)

Conditional Edges

def router(state: State) -> str:
    """Return the name of the next node."""
    if state["count"] > 5:
        return "end_node"
    return "continue_node"

graph.add_conditional_edges(
    "a",
    router,
    {"end_node": END, "continue_node": "b"}
)

Persistence (Checkpointing)

Enable durable execution with checkpointers:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Each thread_id maintains separate state
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [...]}, config)

# Resume from checkpoint
result = app.invoke({"messages": [...]}, config)

Production Checkpointers

# PostgreSQL
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

# SQLite
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("sqlite:///checkpoints.db")

Streaming

app = graph.compile()

# Stream node outputs
for chunk in app.stream({"messages": [...]}):
    print(chunk)

# Stream specific events
for event in app.stream({"messages": [...]}, stream_mode="updates"):
    print(event)

# Stream tokens from LLM
for event in app.stream({"messages": [...]}, stream_mode="messages"):
    print(event)

Human-in-the-Loop

Interrupts

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_node"],  # Pause before this node
)

config = {"configurable": {"thread_id": "123"}}

# Run until interrupt
result = app.invoke({"messages": [...]}, config)

# Review state, then continue
result = app.invoke(None, config)  # Resume from checkpoint

Dynamic Breakpoints

from langgraph.errors import NodeInterrupt

def my_node(state: State):
    if needs_approval(state):
        raise NodeInterrupt("Approval required")
    return {"result": "processed"}

Subgraphs

Compose complex workflows from smaller graphs:

# Define inner graph
inner_graph = StateGraph(State)
inner_graph.add_node("inner_node", inner_fn)
inner_app = inner_graph.compile()

# Use as node in outer graph
outer_graph = StateGraph(State)
outer_graph.add_node("subgraph", inner_app)
outer_graph.add_edge(START, "subgraph")

ReAct Agent Pattern

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

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

graph = StateGraph(MessagesState)
graph.add_node("model", call_model)
graph.add_node("tools", ToolNode(tools))

graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")

app = graph.compile()

Best Practices

  1. Use TypedDict for state - Clear typing, good performance
  2. Define reducers explicitly - Control how state updates are merged
  3. Use checkpointers in production - Enable durability and recovery
  4. Keep nodes focused - Single responsibility, easier testing
  5. Use conditional edges for branching - Clean separation of routing logic
  6. Stream for long operations - Better UX, visible progress

When to Use LangGraph vs LangChain

Use LangGraph when... Use LangChain when...
Complex multi-step workflows Simple agent loops
Custom state management Default message-based state
Fine-grained streaming control Basic streaming needs
Multiple conditional branches Linear tool-calling flows
Heavy customization needed Quick prototyping

Documentation Index

Resource When to Consult
overview.md Getting started, core concepts
quickstart.md First graph tutorial
graph-api.md StateGraph, nodes, edges, state schemas
functional-api.md Functional API alternative
persistence.md Checkpointers, durability
memory.md Memory management patterns
streaming.md Stream modes, token streaming
interrupts.md Human-in-the-loop, breakpoints
durable-execution.md Reliability, recovery
use-subgraphs.md Composing graphs
thinking-in-langgraph.md Mental model, design patterns
workflows-agents.md Workflow vs agent patterns
deploy.md Deployment options
errors/ Error codes and troubleshooting

Syncing Documentation

cd skills/langgraph
bun run scripts/sync-docs.ts