Langgraph

LangGraph Skill

Expert assistance for building LangGraph agents in Python: stateful graphs, checkpointing, human-in-the-loop interrupts, streaming, fault tolerance, and deployment.

Two APIs to choose from:

• Graph API — StateGraph, nodes, edges: explicit DAG control flow
• Functional API — @entrypoint, @task: imperative Python-style workflows

Reference corpus: 30 LangGraph pages in references/llms-txt.md (5.4 MB) and references/llms.md (104 KB). For full content with code, use ../langchain-deepagents/references/llms-full.md.

When to Use This Skill

Activate when:

• Building a StateGraph — defining nodes, edges, and state types with Graph API
• Using the Functional API — @entrypoint / @task for imperative agent logic
• Adding persistence/memory — wiring up InMemorySaver or PostgresSaver checkpointers
• Implementing human-in-the-loop — using interrupt() to pause and resume execution
• Streaming agent output — configuring streaming modes (values, updates, messages, events)
• Handling fault tolerance — per-node retries, timeouts, and error handlers
• Durable execution — replaying from checkpoints, time travel, state updates
• Deploying LangGraph — local server, LangSmith Deployment, langgraph.json config
• Debugging graph state — inspecting thread state, replaying past runs, updating state

Quick Reference

Graph API — define a StateGraph

from typing import TypedDict
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
    foo: str

def my_node(state: State) -> State:
    return {"foo": state["foo"].upper()}

# Build graph
graph = StateGraph(State)
graph.add_node("my_node", my_node)
graph.add_edge(START, "my_node")

# Compile with checkpointer for persistence
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Invoke with thread_id for stateful conversations
result = app.invoke({"foo": "hello"}, config={"configurable": {"thread_id": "1"}})

Separate input/output state types

from typing import TypedDict
from langgraph.graph import StateGraph

class InputState(TypedDict):
    user_input: str

class OutputState(TypedDict):
    graph_output: str

class OverallState(TypedDict):
    foo: str
    user_input: str
    graph_output: str

# Graph sees OverallState internally; caller sees InputState/OutputState
graph = StateGraph(OverallState, input=InputState, output=OutputState)

Functional API — @entrypoint and @task

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy

@task(retry=RetryPolicy(max_attempts=3))
def fetch_data(query: str) -> str:
    return call_api(query)

@entrypoint(checkpointer=InMemorySaver())
def my_workflow(inputs: dict) -> dict:
    data = fetch_data(inputs["query"]).result()
    return {"output": process(data)}

Human-in-the-loop with interrupt

from langgraph.types import interrupt

def review_node(state: State) -> State:
    # Pauses here and returns control to caller
    human_decision = interrupt({"question": "Approve this action?", "data": state["action"]})
    if human_decision == "approve":
        return {"approved": True}
    return {"approved": False}

# Resume by invoking with Command
from langgraph.types import Command
app.invoke(Command(resume="approve"), config={"configurable": {"thread_id": "1"}})

Streaming

# Stream state values after each node
for chunk in app.stream({"foo": "hello"}, config={"configurable": {"thread_id": "1"}},
                         stream_mode="values"):
    print(chunk)

# Stream only updates (deltas)
for chunk in app.stream(inputs, config=config, stream_mode="updates"):
    print(chunk)

# Stream LLM tokens (messages mode)
for chunk in app.stream(inputs, config=config, stream_mode="messages"):
    print(chunk)

Fault tolerance — per-node retries and timeouts

from langgraph.graph import StateGraph
from langgraph.types import RetryPolicy

graph = StateGraph(State)
graph.add_node(
    "flaky_node",
    my_node,
    retry=RetryPolicy(max_attempts=3, backoff_factor=2.0),
)

Time travel — replay from checkpoint

# Get thread history
history = list(app.get_state_history(config={"configurable": {"thread_id": "1"}}))

# Pick a prior checkpoint and replay from it
past_checkpoint = history[2].config
result = app.invoke(None, config=past_checkpoint)

# Update state at a checkpoint before replaying
app.update_state(config=past_checkpoint, values={"foo": "corrected"})

Key LangGraph Concepts

Reference Files

LangGraph Doc Index (Python)

Key pages at docs.langchain.com/oss/python/langgraph/:

Updating

skill-seekers scrape --url "https://docs.langchain.com/oss/python/langgraph/overview" --name langgraph