Advanced LangGraph features

You already know how to build a graph, stream its output, and use checkpointing. Because LangGraph saves a snapshot after every step, you can pause a run for human approval, recover from crashes, and even rewind a conversation to an earlier point. In this topic, we explore these and more capabilities one by one.

Long-term memory

LangGraph has two kinds of memory. Short-term memory is the conversation history in a single thread. Thecheckpointer we built earlier handles this. On the other hand, long-term memory is information that can be used by any thread — user preferences, facts, and learned habits.

In LangGraph, you create a store for this information:

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

This information is stored in a namespace:

ns = ("user", "hyperskill")

The memory itself is a key-value where the key is a unique identifier for the memory in that namespace and the value is a dictionary containing the data:

key = "preferences"
value = {"language": "python"}

Finally, use store.put to save the memory in the store:

store.put(namespace=ns, key=key, value=value)

You can retrieve memories from a particular namespace using store.search:

print(store.search(ns)[-1].dict())

# output: {'namespace': ['users', 'hyperskill'], 'key': 'preferences', 'value': {'language': 'python'}, 'created_at': '2026-07-26T10:47:12.640917+00:00', 'updated_at': '2026-07-26T10:47:12.640922+00:00', 'score': None}

To fetch a specific memory, use store.get. You can also use semantic search for memories to fetch memories based on meaning. For this to work, you need to configure your store with an embedding index. Without an index, you can retrieve all memories in a namespace or access a specific memory by its key.

Each result is a store item. You can access the stored dictionary through its value attribute:

memories = store.search(ns)

preferred_language = memories[-1].value["language"]
print(preferred_language)

# Output: python

Once you have created the store, pass it to .compile:

graph = builder.compile(store=store)

Compiling the graph with the store makes it available to graph nodes through the Runtime object. However, storing a memory does not automatically make the model aware of it. You must retrieve the memory and include it in the model call.

First, define the runtime context. This context identifies the user whose memories should be retrieved:

from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

Next, let’s expand the chatbot from the previous topic to use this context:

from langchain_core.messages import SystemMessage
from langgraph.runtime import Runtime
    

def chatbot(state: State, runtime: Runtime[Context]):
    user_id = runtime.context.user_id
    namespace = ("users", user_id)

    memories = runtime.store.search(namespace)

    memory_text = "\n".join(
        f"- Preferred programming language: {item.value['language']}"
        for item in memories
        if "language" in item.value
    )

    system_message = SystemMessage(
        content=(
            "You are a helpful assistant. Use the following stored user "
            "information when it is relevant:\n\n"
            f"{memory_text or '- No stored information'}"
        )
    )

    response = model.invoke([system_message, *state["messages"]])
    return {"messages": [response]}

The retrieved memories are converted into text and added to a system message. This gives the model access to the stored information when it generates its response. When creating the graph, provide the context schema:

builder = StateGraph(State, context_schema=Context)
# add nodes and edges as before

Finally, pass the appropriate user ID through the runtime context when invoking the graph:

result = graph.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "What programming language do I prefer?",
            }
        ]
    },
    context={"user_id": "hyperskill"},
)

print(result["messages"][-1].content)

# output: You prefer **Python**.

InMemoryStore stores its data in memory, so all information is lost when the application stops. For persistent long-term memory, use a database-backed store.

Interrupts: human-in-the-loop

Some actions should not happen without human oversight. For example, an agent may need approval before sending an email, deleting a file, or making a purchase. LangGraph supports this with interrupts.

Calling interrupt() inside a node pauses the graph and returns a value to the caller:

from typing_extensions import TypedDict
from langgraph.types import interrupt

class ReviewState(TypedDict):
    generated_text: str
    status: str

def human_review(state: ReviewState):
    decision = interrupt(
        {
            "question": "Do you approve this response?",
            "content": state["generated_text"],
        }
    )

    if decision == "yes":
        return {"status": "approved"}

    return {"status": "rejected"}

The value passed to interrupt() should contain enough information for a user interface or reviewer to understand the decision. It must also be JSON-serializable so that LangGraph can save it in a checkpoint.

Add the review node to a graph as usual:

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END

builder = StateGraph(ReviewState)
builder.add_node("human_review", human_review)
builder.add_edge(START, "human_review")
builder.add_edge("human_review", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Interrupts require a checkpointer and a thread ID. The thread ID allows LangGraph to locate the paused execution later:

config = {"configurable": {"thread_id": "review-001"}}

result = graph.invoke(
    {
        "generated_text": "Recursion is a function calling itself.",
        "status": "waiting",
    },
    config=config,
)

print(result["__interrupt__"])

At this point, the graph has not reached END. Its state has been saved, and the interrupt information is available through the special __interrupt__ key.

To continue, invoke the graph again with a Command:

from langgraph.types import Command

result = graph.invoke(
    Command(resume="yes"),
    config=config,
)

print(result["status"])

# output: approved

The value supplied to Command(resume=...) becomes the return value of the original interrupt() call. In this example, decision receives "yes" and the node returns an approved status.

When a graph resumes, the interrupted node starts again from the beginning rather than continuing from the exact Python line where it paused. Therefore, code before interrupt() may run more than once. Any side effects in that part of the node should be idempotent, meaning that repeating them produces the same result without causing duplicate operations. For example, prefer updating a database record by a stable identifier over blindly inserting a new record.

Subgraphs

As applications grow, putting every operation in one graph can make the workflow difficult to understand. A subgraph is a graph used as a node inside another graph. It lets you divide a large application into smaller workflows that can be developed and tested independently.

Suppose part of an application prepares and reviews a draft:

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

class DocumentState(TypedDict):
    topic: str
    draft: str

def create_draft(state: DocumentState):
    return {"draft": f"A short introduction to {state['topic']}."}

def improve_draft(state: DocumentState):
    return {"draft": state["draft"] + " It includes practical examples."}

subgraph_builder = StateGraph(DocumentState)
subgraph_builder.add_node("create_draft", create_draft)
subgraph_builder.add_node("improve_draft", improve_draft)
subgraph_builder.add_edge(START, "create_draft")
subgraph_builder.add_edge("create_draft", "improve_draft")
subgraph_builder.add_edge("improve_draft", END)

drafting_subgraph = subgraph_builder.compile()

The compiled subgraph can then be added directly to a parent graph:

def publish(state: DocumentState):
    print(state["draft"])
    return {}

parent_builder = StateGraph(DocumentState)
parent_builder.add_node("drafting", drafting_subgraph)
parent_builder.add_node("publish", publish)
parent_builder.add_edge(START, "drafting")
parent_builder.add_edge("drafting", "publish")
parent_builder.add_edge("publish", END)

graph = parent_builder.compile()

This direct approach works when the parent and subgraph share the state keys the subgraph needs. The subgraph reads from and writes to the same state channels as the parent.

If their state schemas are different, wrap the subgraph in a function that converts the parent state into the subgraph's input and maps its result back:

def call_drafting_subgraph(state: ParentState):
    result = drafting_subgraph.invoke({"topic": state["subject"], "draft": ""})
    return {"article": result["draft"]}

By default, a subgraph used inside a persistent parent graph can inherit the parent's checkpointer. You can also configure subgraph-specific persistence when the subgraph needs its own thread-level memory or independently inspectable checkpoints.

Fault tolerance

Model calls, network requests, and external services can fail temporarily. Because a graph compiled with a checkpointer saves state at each completed step, LangGraph can resume from the last successful checkpoint rather than rerunning the entire workflow.

For transient errors, you can attach a retry policy to a node:

from langgraph.types import RetryPolicy

builder.add_node(
    "call_external_service",
    call_external_service,
    retry_policy=RetryPolicy(
        max_attempts=3,
        retry_on=ConnectionError,
    ),
)

If call_external_service raises ConnectionError, LangGraph tries the node again, up to the configured maximum number of attempts. Errors that do not match retry_on are not retried by this policy. This distinction is useful because temporary connection problems may disappear, while validation and programming errors usually require a code or input change.

If a run still fails, its completed checkpoints remain available. After fixing the problem, invoke the graph again for the same thread without providing new input:

config = {"configurable": {"thread_id": "job-001"}}

try:
    graph.invoke(initial_state, config=config)
except ConnectionError:
    print("The workflow could not reach the external service.")

# After the service or application has been fixed:
result = graph.invoke(None, config=config)

LangGraph restores the saved state and continues from the pending step. Nodes that completed and were checkpointed do not need to run again.

However, a node itself is the unit of checkpointing. If a node fails halfway through, it starts again from the beginning when retried or resumed. Operations with side effects should therefore be idempotent. Alternatively, split a large node into several smaller nodes so that each successfully completed operation gets its own checkpoint.

For production applications, use a persistent checkpointer backed by a database. InMemorySaver is useful for local development, but its checkpoints disappear when the process stops.

Time travel

Since a checkpointer records state after each graph step, you can inspect previous states and continue execution from an earlier checkpoint. LangGraph calls this time travel.

First, run a graph with a thread ID:

config = {"configurable": {"thread_id": "conversation-001"}}

result = graph.invoke(initial_state, config=config)

Then retrieve the thread's checkpoint history:

history = list(graph.get_state_history(config))

for snapshot in history:
    print("Values:", snapshot.values)
    print("Next nodes:", snapshot.next)
    print("Checkpoint:", snapshot.config["configurable"]["checkpoint_id"])

get_state_history() returns snapshots in reverse chronological order, with the most recent state first. Each snapshot contains the state values, the nodes scheduled to run next, metadata about the step, and a configuration identifying that checkpoint.

To replay the graph, select a snapshot and invoke the graph with its configuration:

checkpoint = next(
    snapshot
    for snapshot in history
    if snapshot.next == ("chatbot",)
)

result = graph.invoke(None, checkpoint.config)

LangGraph does not rerun the steps that occurred before the selected checkpoint. It reuses their saved results and executes the nodes listed in snapshot.next and everything that follows.

You can also create a fork by changing the state at a previous checkpoint:

fork_config = graph.update_state(
    checkpoint.config,
    {
        "messages": [
            {
                "role": "user",
                "content": "Explain the same idea using an analogy.",
            }
        ]
    },
)

result = graph.invoke(None, fork_config)

update_state() creates a new checkpoint rather than modifying the old one. Continuing from its returned configuration produces a new branch of execution, while the original history remains available. This is useful for editing a past message, testing another decision, or comparing several model responses from the same point in a workflow.

Conclusion

LangGraph goes beyond basic workflow orchestration by giving agents continuity, structure, and resilience. Long-term memory carries useful knowledge across conversations, interrupts create space for human judgment, and subgraphs keep expanding systems organized. Fault-tolerance mechanisms help interrupted runs recover safely, while state forking lets you revisit checkpoints and branch into alternative outcomes. With these tools, you can design agentic systems that remain adaptable and dependable as their responsibilities grow.