LangGraph

https://docs.langchain.com/oss/python/langgraph/
Admin
LangGraph is a low-level orchestration framework and runtime for building, managing, and deploying...
Tokens:219,947
Snippets:1,641
Trust Score:10
Update:3 days ago
Show doc for...
Tokens:
Raw
### Sync Postgres checkpointer example with setup

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Complete sync example using PostgresSaver. Call checkpointer.setup() the first time to create necessary tables. The graph streams events for two user messages to demonstrate memory.

```python
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres import PostgresSaver  

model = init_chat_model(model="claude-haiku-4-5-20251001")

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # checkpointer.setup()

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

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(checkpointer=checkpointer)

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

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,
        version="v3",
    )
    for snapshot in stream.values:
        print(snapshot)

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config,
        version="v3",
    )
    for snapshot in stream.values:
        print(snapshot)
```

--------------------------------

### Install dependencies with pip

Source: https://docs.langchain.com/oss/python/langgraph/studio

Installs the required Python packages for the example agent from the project root directory.

```shell
pip install langchain langchain-openai
```

--------------------------------

### Async Postgres checkpointer example with setup

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Async version using AsyncPostgresSaver. Call await checkpointer.setup() the first time. Streams events asynchronously and prints tokens.

```python
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver  

model = init_chat_model(model="claude-haiku-4-5-20251001")

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # await checkpointer.setup()

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

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(checkpointer=checkpointer)

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

    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,
        version="v3",
    )
    async for message in stream.messages:
        async for token in message.text:
            print(token, end="", flush=True)

    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config,
        version="v3",
    )
    async for message in stream.messages:
        async for token in message.text:
            print(token, end="", flush=True)
```

--------------------------------

### Sync OracleStore and OracleSaver setup with LangGraph memory

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Use this sync example to configure OracleStore and OracleSaver with an Oracle AI Database. Call store.setup() and checkpointer.setup() the first time you use them. The example requires an Oracle AI Database instance with Oracle AI Vector Search for semantic search.

```python
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.store.base import BaseStore
from langgraph_oracledb.checkpoint.oracle import OracleSaver
from langgraph_oracledb.store.oracle import OracleStore  

model = init_chat_model(model="claude-haiku-4-5-20251001")
embeddings = init_embeddings("openai:text-embedding-3-small")

DB_URI = "user/password@localhost:1521/FREEPDB1"

with (
    OracleStore.from_conn_string(
        DB_URI,
        index={"embed": embeddings, "dims": 1536},
    ) as store,
    OracleSaver.from_conn_string(DB_URI) as checkpointer,
):
    store.setup()
    checkpointer.setup()

    def call_model(
        state: MessagesState,
        config: RunnableConfig,
        *,
        store: BaseStore,
    ):
        user_id = config["configurable"]["user_id"]
        namespace = ("memories", user_id)
        memories = store.search(namespace, query=str(state["messages"][-1].content))
        info = "\n".join([d.value["data"] for d in memories])
        system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

        # Store new memories if the user asks the model to remember
        last_message = state["messages"][-1]
        if "remember" in last_message.content.lower():
            memory = "User name is Bob"
            store.put(namespace, str(uuid7()), {"data": memory})

        response = model.invoke(
            [{"role": "system", "content": system_msg}] + state["messages"]
        )
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

    config = {
        "configurable": {
            "thread_id": "1",
            "user_id": "1",
        }
    }
    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
        config,
        version="v3",
    )
    for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

    config = {
        "configurable": {
            "thread_id": "2",
            "user_id": "1",
        }
    }

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "what is my name?"}]},
        config,
        version="v3",
    )
    for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()
```

--------------------------------

### Set a conditional entry point using START

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Use add_conditional_edges from the virtual START node to dynamically choose the starting node(s) based on custom logic.

```python
from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)
```

--------------------------------

### Install pytest

Source: https://docs.langchain.com/oss/python/langgraph/test

Install or upgrade pytest using pip. Required before running the test examples.

```bash
$ pip install -U pytest
```

--------------------------------

### Build a sequential graph with add_node and add_edge

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Create a sequence of steps by adding nodes and connecting them with edges from START. This example uses StateGraph and the add_node and add_edge methods.

```python
from langgraph.graph import START, StateGraph

builder = StateGraph(State)

# Add nodes
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)

# Add edges
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
```

--------------------------------

### Install LangGraph SDK

Source: https://docs.langchain.com/oss/python/langgraph/deploy

Install the LangGraph SDK to interact with your deployed agent programmatically.

```bash
pip install langgraph-sdk
```

--------------------------------

### Clone Agent Chat UI repository for local development

Source: https://docs.langchain.com/oss/python/langgraph/ui

Clones the agent-chat-ui repository, then installs dependencies and starts the dev server with pnpm.

```shell
# Clone the repository
git clone https://github.com/langchain-ai/agent-chat-ui.git
cd agent-chat-ui

# Install dependencies and start
pnpm install
pnpm dev
```

--------------------------------

### Install HuggingFace chat model integration with uv

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain HuggingFace integration package using uv.

```bash
uv add "langchain[huggingface]"
```

--------------------------------

### Create Agent Chat UI project with npx create-agent-chat-app

Source: https://docs.langchain.com/oss/python/langgraph/ui

Scaffolds a new Agent Chat UI project named my-chat-ui, then installs dependencies and starts the dev server with pnpm.

```shell
# Create a new Agent Chat UI project
npx create-agent-chat-app --project-name my-chat-ui
cd my-chat-ui

# Install dependencies and start
pnpm install
pnpm dev
```

--------------------------------

### Install OpenAI chat model integration with uv

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain OpenAI integration package using uv.

```bash
uv add "langchain[openai]"
```

--------------------------------

### Install Postgres checkpointer dependencies

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Install the required packages for using PostgresSaver with psycopg.

```bash
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
```

--------------------------------

### Full approval workflow example with StateGraph and InMemorySaver

Source: https://docs.langchain.com/oss/python/langgraph/interrupts

Complete example showing an approval node, proceed and cancel nodes, a state graph with InMemorySaver checkpointer, and resuming with a decision. Use a durable checkpointer in production.

```python
from typing import Literal, Optional, TypedDict

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


class ApprovalState(TypedDict):
    action_details: str
    status: Optional[Literal["pending", "approved", "rejected"]]


def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
    # Expose details so the caller can render them in a UI
    decision = interrupt(
        {
            "question": "Approve this action?",
            "details": state["action_details"],
        }
    )

    # Route to the appropriate node after resume
    return Command(goto="proceed" if decision else "cancel")


def proceed_node(state: ApprovalState):
    return {"status": "approved"}


def cancel_node(state: ApprovalState):
    return {"status": "rejected"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("proceed", proceed_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("proceed", END)
builder.add_edge("cancel", END)

# Use a more durable checkpointer in production
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-123"}}
initial = graph.stream_events(
    {"action_details": "Transfer $500", "status": "pending"},
    config=config,
    version="v3",
)
_ = initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'question': ..., 'details': ...}),)

# Resume with the decision; True routes to proceed, False to cancel
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
print(resumed.output["status"])
```

--------------------------------

### Install dependencies with uv sync

Source: https://docs.langchain.com/oss/python/langgraph/local-server

Installs the app's dependencies using uv sync. Run from the root of your new LangGraph app.

```shell
cd path/to/your/app
uv sync
```

--------------------------------

### Install langchain-openrouter with pip

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Use pip to install the langchain-openrouter package.

```bash
pip install -U "langchain-openrouter"
```

--------------------------------

### Install Google Gemini chat model integration with uv

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain Google GenAI integration package using uv.

```bash
uv add "langchain[google-genai]"
```

--------------------------------

### Install Oracle Store Dependencies

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Install the required packages to use Oracle store with LangGraph.

```bash
pip install -U langgraph langgraph-oracledb langchain-openai
```

--------------------------------

### Install LangGraph CLI with uv

Source: https://docs.langchain.com/oss/python/langgraph/local-server

Installs the LangGraph CLI with in-memory server support using uv. Requires Python 3.11 or higher.

```shell
# Python >= 3.11 is required.
uv add "langgraph-cli[inmem]"
```

--------------------------------

### Example output of summarized conversation

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

This is the printed output from the full example, showing the AI response and the running summary.

```text
================================== Ai Message ==================================

From our conversation, I can see that you introduced yourself as Bob. That's the name you shared with me when we began talking.

Summary: In this conversation, I was introduced to Bob, who then asked me to write a poem about cats. I composed a poem titled "The Mystery of Cats" that captured cats' graceful movements, independent nature, and their special relationship with humans. Bob then requested a similar poem about dogs, so I wrote "The Joy of Dogs," which highlighted dogs' loyalty, enthusiasm, and loving companionship. Both poems were written in a similar style but emphasized the distinct characteristics that make each pet special.
```

--------------------------------

### Full example: tool approval workflow with SQLite checkpointer

Source: https://docs.langchain.com/oss/python/langgraph/interrupts

This complete example builds a state graph that uses the send_email tool with an interrupt. It uses SqliteSaver for checkpointing and demonstrates streaming events, inspecting interrupts, and resuming with approval and edited arguments.

```python
import sqlite3
import operator
from typing import TypedDict, Annotated, Literal
from langchain.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import AnyMessage, SystemMessage, ToolMessage


class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]


@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""

    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?",
    })

    if response.get("action") == "approve":
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)

        # Actually send the email (your implementation here)
        print(f"[send_email] to={final_to} subject={final_subject} body={final_body}")
        return f"Email sent to {final_to}"

    return "Email cancelled by user"


model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email])
tools_by_name = {"send_email": send_email}


def agent_node(state: AgentState):
    # LLM may decide to call the tool; interrupt pauses before sending
    result = model.invoke(state["messages"])
    return {"messages": [result]}

def tool_node(state: AgentState):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

def should_continue(state: AgentState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    if last_message.tool_calls:
        return "tool_node"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tool_node", tool_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tool_node", END])  # Routes to "tools" or END
builder.add_edge("tool_node", "agent")  # Loop back after tools

checkpointer = SqliteSaver(
    sqlite3.connect("tool-approval.db", check_same_thread=False)
)
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "email-workflow"}}
initial = graph.stream_events(
    {
        "messages": [
            {"role": "user", "content": "Send an email to alice@example.com about the meeting"}
        ]
    },
    config=config,
    version="v3",
)
initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'action': 'send_email', ...}),)

# Resume with approval and optionally edited arguments
resumed = graph.stream_events(
    Command(resume={"action": "approve", "subject": "Updated subject"}),
    config=config,
    version="v3",
)
print(resumed.output["messages"][-1])  # -> Tool result returned by send_email
```

--------------------------------

### Install langchain-openrouter with uv

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Use uv to add the langchain-openrouter package to your project.

```bash
uv add "langchain-openrouter"
```

--------------------------------

### Install dependencies with uv

Source: https://docs.langchain.com/oss/python/langgraph/studio

Alternative command to install the required Python packages using uv from the project root directory.

```shell
uv add langchain langchain-openai
```

--------------------------------

### add_edge from START

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Sets the entry point of the graph by adding an edge from the virtual START node to the first node to execute.

```APIDOC
## add_edge from START

### Description
Sets the entry point of the graph by adding an edge from the virtual START node to the first node to execute.

### Method
`add_edge(START, target)`

### Parameters
- **START** (constant) - Required - The virtual start node imported from `langgraph.graph`.
- **target** (str) - Required - The name of the first node to execute.

### Example
```python
from langgraph.graph import START

graph.add_edge(START, "node_a")
```
```

--------------------------------

### set_entry_point(node)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Defines the first node the graph will execute. It is equivalent to `builder.add_edge(START, node)`. Both methods are valid but `add_edge(START, ...)` is the recommended modern syntax.

```APIDOC
## set_entry_point(node)

### Description
Defines the first node the graph will execute. It is equivalent to `builder.add_edge(START, node)`. Both methods are valid but `add_edge(START, ...)` is the recommended modern syntax.

### Parameters
- **node** (str) - Required - The name of the node to set as the entry point.

### Usage
```python
builder.set_entry_point("node_a")
```
```

--------------------------------

### Install AWS Bedrock chat model integration with uv

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain AWS integration package using uv.

```bash
uv add "langchain[aws]"
```

--------------------------------

### Install Redis checkpointer package

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Install the langgraph-checkpoint-redis package to use the Redis checkpointer. You must call checkpointer.setup() the first time you use it.

```bash
pip install -U langgraph langgraph-checkpoint-redis
```

--------------------------------

### Install Anthropic chat model integration with uv

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain Anthropic integration package using uv.

```bash
uv add "langchain[anthropic]"
```

--------------------------------

### Full LangGraph Functional API agent example

Source: https://docs.langchain.com/oss/python/langgraph/quickstart

Complete runnable example: defines tools and model, the call_llm model node, the call_tool tool node, and the agent entrypoint, then invokes it with stream_events.

```python
# Step 1: Define tools and model

from langchain.tools import tool
from langchain.chat_models import init_chat_model


model = init_chat_model(
    "claude-sonnet-4-6",
    temperature=0
)


# Define tools
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# Augment the LLM with tools
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

from langgraph.graph import add_messages
from langchain.messages import (
    SystemMessage,
    HumanMessage,
    ToolCall,
)
from langchain_core.messages import BaseMessage
from langgraph.func import entrypoint, task


# Step 2: Define model node

@task
def call_llm(messages: list[BaseMessage]):
    """LLM decides whether to call a tool or not"""
    return model_with_tools.invoke(
        [
            SystemMessage(
                content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
            )
        ]
        + messages
    )


# Step 3: Define tool node

@task
def call_tool(tool_call: ToolCall):
    """Performs the tool call"""
    tool = tools_by_name[tool_call["name"]]
    return tool.invoke(tool_call)


# Step 4: Define agent

@entrypoint()
def agent(messages: list[BaseMessage]):
    model_response = call_llm(messages).result()

    while True:
        if not model_response.tool_calls:
            break

        # Execute tools
        tool_result_futures = [
            call_tool(tool_call) for tool_call in model_response.tool_calls
        ]
        tool_results = [fut.result() for fut in tool_result_futures]
        messages = add_messages(messages, [model_response, *tool_results])
        model_response = call_llm(messages).result()

    messages = add_messages(messages, model_response)
    return messages

# Invoke
messages = [HumanMessage(content="Add 3 and 4.")]
stream = agent.stream_events(messages, version="v3")
for snapshot in stream.values:
    print(snapshot)
    print("\n")
```

--------------------------------

### Install Google Gemini chat model integration with pip

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain Google GenAI integration package using pip.

```bash
pip install -U "langchain[google-genai]"
```

--------------------------------

### Full example: Summarize messages with SummarizationNode

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Complete example using SummarizationNode from langmem to summarize messages, with a running summary stored in the context field. Invoke the graph with a thread_id to see the summary and final response.

```python
from typing import Any, TypedDict

from langchain.chat_models import init_chat_model
from langchain.messages import AnyMessage
from langchain_core.messages.utils import count_tokens_approximately
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langmem.short_term import SummarizationNode, RunningSummary  

model = init_chat_model("claude-sonnet-4-6")
summarization_model = model.bind(max_tokens=128)

class State(MessagesState):
    context: dict[str, RunningSummary]

class LLMInputState(TypedDict):
    summarized_messages: list[AnyMessage]
    context: dict[str, RunningSummary]

summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately,
    model=summarization_model,
    max_tokens=256,
    max_tokens_before_summary=256,
    max_summary_tokens=128,
)

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

checkpointer = InMemorySaver()
builder = StateGraph(State)
builder.add_node(call_model)
builder.add_node("summarize", summarization_node)
builder.add_edge(START, "summarize")
builder.add_edge("summarize", "call_model")
graph = builder.compile(checkpointer=checkpointer)

# Invoke the graph
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"messages": "hi, my name is bob"}, config)
graph.invoke({"messages": "write a short poem about cats"}, config)
graph.invoke({"messages": "now do the same but for dogs"}, config)
final_response = graph.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()
print("\nSummary:", final_response["context"]["running_summary"].summary)
```

--------------------------------

### Install LangGraph with uv

Source: https://docs.langchain.com/oss/python/langgraph/install

Adds the base LangGraph package to a project using uv.

```shell
uv add langgraph
```

--------------------------------

### Install LangGraph CLI with in-memory server

Source: https://docs.langchain.com/oss/python/langgraph/studio

Installs the LangGraph CLI with the inmem extra, which provides a local development server for connecting your agent to Studio. Requires Python 3.11 or higher.

```shell
# Python >= 3.11 is required.
pip install --upgrade "langgraph-cli[inmem]"
```

--------------------------------

### Install LangChain with uv

Source: https://docs.langchain.com/oss/python/langgraph/install

Adds LangChain to a project using uv. Requires Python 3.10+.

```shell
uv add langchain
# Requires Python 3.10+
```

--------------------------------

### Install Oracle Checkpointer for LangGraph

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Install the langgraph-oracledb package to use the Oracle checkpointer. Requires an Oracle AI Database instance (local container or OCI Autonomous Database) and calling checkpointer.setup() on first use.

```bash
pip install -U langgraph langgraph-oracledb
```

--------------------------------

### Complete LangGraph review and edit example with MemorySaver

Source: https://docs.langchain.com/oss/python/langgraph/interrupts

A full runnable example that defines a graph with a review node, compiles it with a MemorySaver checkpointer, and demonstrates pausing and resuming with edited content. The initial stream shows the interrupt payload, and the resumed stream outputs the edited text.

```python
from typing import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class ReviewState(TypedDict):
    generated_text: str


def review_node(state: ReviewState):
    # Ask a reviewer to edit the generated content
    updated = interrupt(
        {
            "instruction": "Review and edit this content",
            "content": state["generated_text"],
        }
    )
    return {"generated_text": updated}


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

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

config = {"configurable": {"thread_id": "review-42"}}
initial = graph.stream_events(
    {"generated_text": "Initial draft"}, config=config, version="v3"
)
_ = initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'instruction': ..., 'content': ...}),)

# Resume with the edited text from the reviewer
final_state = graph.stream_events(
    Command(resume="Improved draft after review"),
    config=config,
    version="v3",
)
print(final_state.output["generated_text"])  # -> "Improved draft after review"
```

--------------------------------

### Example output of graph.nodes

Source: https://docs.langchain.com/oss/python/langgraph/pregel

The output shows a dictionary mapping node names to PregelNode instances.

```python
{'__start__': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1810>,
 'write_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba14d0>,
 'score_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1710>}
```

--------------------------------

### Hello world example with StateGraph and MessagesState

Source: https://docs.langchain.com/oss/python/langgraph/overview

Create a simple graph with a mock LLM node that returns a fixed message. Invoke the compiled graph with a user message to see the response.

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

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()

graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
```

--------------------------------

### Add memory to LangGraph agent using AsyncOracleStore and AsyncOracleSaver

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Use this example to set up an Oracle-backed store with vector search and a checkpointer, then run a graph that remembers user info across threads. Requires an Oracle connection string and an embedding model; call store.setup() and checkpointer.setup() before use.

```python
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.store.base import BaseStore
from langgraph_oracledb.checkpoint.oracle import AsyncOracleSaver
from langgraph_oracledb.store.oracle import AsyncOracleStore  

model = init_chat_model(model="claude-haiku-4-5-20251001")
embeddings = init_embeddings("openai:text-embedding-3-small")

DB_URI = "user/password@localhost:1521/FREEPDB1"

async with (
    AsyncOracleStore.from_conn_string(
        DB_URI,
        index={"embed": embeddings, "dims": 1536},
    ) as store,
    AsyncOracleSaver.from_conn_string(DB_URI) as checkpointer,
):
    await store.setup()
    await checkpointer.setup()

    async def call_model(
        state: MessagesState,
        config: RunnableConfig,
        *,
        store: BaseStore,
    ):
        user_id = config["configurable"]["user_id"]
        namespace = ("memories", user_id)
        memories = await store.asearch(namespace, query=str(state["messages"][-1].content))
        info = "\n".join([d.value["data"] for d in memories])
        system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

        # Store new memories if the user asks the model to remember
        last_message = state["messages"][-1]
        if "remember" in last_message.content.lower():
            memory = "User name is Bob"
            await store.aput(namespace, str(uuid7()), {"data": memory})

        response = await model.ainvoke(
            [{"role": "system", "content": system_msg}] + state["messages"]
        )
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

    config = {
        "configurable": {
            "thread_id": "1",
            "user_id": "1",
        }
    }
    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
        config,
        version="v3",
    )
    async for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

    config = {
        "configurable": {
            "thread_id": "2",
            "user_id": "1",
        }
    }

    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "what is my name?"}]},
        config,
        version="v3",
    )
    async for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()
```

--------------------------------

### Project structure after setup

Source: https://docs.langchain.com/oss/python/langgraph/studio

Shows the expected directory layout after creating the agent, .env, and langgraph.json files.

```plaintext
my-app/
├── src
│   └── agent.py
├── .env
└── langgraph.json
```

--------------------------------

### Build and invoke a LangGraph tool-calling agent

Source: https://docs.langchain.com/oss/python/langgraph/quickstart

Complete quickstart that defines multiply/add/divide tools, binds them to a chat model, builds a StateGraph with llm_call and tool_node, and invokes it with "Add 3 and 4.". Requires langchain, langgraph, typing_extensions, and IPython for graph display.

```python
# Step 1: Define tools and model

from langchain.tools import tool
from langchain.chat_models import init_chat_model


model = init_chat_model(
    "claude-sonnet-4-6",
    temperature=0
)


# Define tools
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# Augment the LLM with tools
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

# Step 2: Define state

from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator


class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

# Step 3: Define model node
from langchain.messages import SystemMessage


def llm_call(state: MessagesState):
    """LLM decides whether to call a tool or not"""

    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }


# Step 4: Define tool node

from langchain.messages import ToolMessage


def tool_node(state: MessagesState):
    """Performs the tool call"""

    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

# Step 5: Define logic to determine whether to end

from typing import Literal
from langgraph.graph import StateGraph, START, END


# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""

    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END

# Step 6: Build agent

# Build workflow
agent_builder = StateGraph(MessagesState)

# Add nodes
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# Add edges to connect nodes
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

# Compile the agent
agent = agent_builder.compile()


from IPython.display import Image, display
# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# Invoke
from langchain.messages import HumanMessage
messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()
```

--------------------------------

### Add edge from START node

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Use the START node to specify which node should be called first in the graph. Import START from langgraph.graph and add an edge from START to the first node.

```python
from langgraph.graph import START

graph.add_edge(START, "node_a")
```

--------------------------------

### add_conditional_edges from START

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Sets a conditional entry point by adding conditional edges from the virtual START node. The routing function determines which node(s) to start with.

```APIDOC
## add_conditional_edges from START

### Description
Sets a conditional entry point by adding conditional edges from the virtual START node. The routing function determines which node(s) to start with.

### Method
`add_conditional_edges(START, routing_function, path_map=None)`

### Parameters
- **START** (constant) - Required - The virtual start node imported from `langgraph.graph`.
- **routing_function** (callable) - Required - A function that takes the current state and returns a value indicating the next node(s).
- **path_map** (dict) - Optional - A dictionary mapping the routing function's output to node names.

### Example
```python
from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)
```

### Example with path_map
```python
from langgraph.graph import START

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})
```
```

--------------------------------

### Create .env file with LANGSMITH_API_KEY

Source: https://docs.langchain.com/oss/python/langgraph/local-server

Copy the contents of .env.example into a new .env file and fill in the necessary API keys, such as LANGSMITH_API_KEY.

```bash
LANGSMITH_API_KEY=lsv2...
```

--------------------------------

### Sync Redis Store Memory Example

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Use RedisStore and RedisSaver with synchronous methods. Call store.setup() and checkpointer.setup() the first time using Redis store.

```python
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis import RedisSaver
from langgraph.store.redis import RedisStore  
from langgraph.runtime import Runtime  

model = init_chat_model(model="claude-haiku-4-5-20251001")

@dataclass
class Context:
    user_id: str

def call_model(
    state: MessagesState,
    runtime: Runtime[Context],
):
    user_id = runtime.context.user_id  
    namespace = ("memories", user_id)
    memories = runtime.store.search(namespace, query=str(state["messages"][-1].content))
    info = "\n".join([d.value["data"] for d in memories])
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

    # Store new memories if the user asks the model to remember
    last_message = state["messages"][-1]
    if "remember" in last_message.content.lower():
        memory = "User name is Bob"
        runtime.store.put(namespace, str(uuid7()), {"data": memory})

    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}

DB_URI = "redis://localhost:6379"

with (
    RedisStore.from_conn_string(DB_URI) as store,
    RedisSaver.from_conn_string(DB_URI) as checkpointer,
):
    store.setup()
    checkpointer.setup()

    builder = StateGraph(MessagesState, context_schema=Context)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

    config = {"configurable": {"thread_id": "1"}}
    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
        config,
        version="v3",
        context=Context(user_id="1"),
    )
    for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

    config = {"configurable": {"thread_id": "2"}}
    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "what is my name?"}]},
        config,
        version="v3",
        context=Context(user_id="1"),
    )
    for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()
```

--------------------------------

### Async Redis Store Memory Example

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Use AsyncRedisStore and AsyncRedisSaver with async methods. Call store.setup() and checkpointer.setup() the first time using Redis store.

```python
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis.aio import AsyncRedisSaver
from langgraph.store.redis.aio import AsyncRedisStore  
from langgraph.runtime import Runtime  

model = init_chat_model(model="claude-haiku-4-5-20251001")

@dataclass
class Context:
    user_id: str

async def call_model(
    state: MessagesState,
    runtime: Runtime[Context],
):
    user_id = runtime.context.user_id  
    namespace = ("memories", user_id)
    memories = await runtime.store.asearch(namespace, query=str(state["messages"][-1].content))
    info = "\n".join([d.value["data"] for d in memories])
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

    # Store new memories if the user asks the model to remember
    last_message = state["messages"][-1]
    if "remember" in last_message.content.lower():
        memory = "User name is Bob"
        await runtime.store.aput(namespace, str(uuid7()), {"data": memory})

    response = await model.ainvoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}

DB_URI = "redis://localhost:6379"

async with (
    AsyncRedisStore.from_conn_string(DB_URI) as store,
    AsyncRedisSaver.from_conn_string(DB_URI) as checkpointer,
):
    # await store.setup()
    # await checkpointer.asetup()

    builder = StateGraph(MessagesState, context_schema=Context)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

    config = {"configurable": {"thread_id": "1"}}
    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
        config,
        version="v3",
        context=Context(user_id="1"),
    )
    async for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

    config = {"configurable": {"thread_id": "2"}}
    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "what is my name?"}]},
        config,
        version="v3",
        context=Context(user_id="1"),
    )
    async for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()
```

--------------------------------

### Install MongoDB checkpointer dependencies

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Install the required packages for using the MongoDB checkpointer.

```bash
pip install -U pymongo langgraph langgraph-checkpoint-mongodb
```

--------------------------------

### Install LangGraph with pip

Source: https://docs.langchain.com/oss/python/langgraph/install

Installs or upgrades the base LangGraph package using pip.

```shell
pip install -U langgraph
```

--------------------------------

### Sample output from langgraph dev

Source: https://docs.langchain.com/oss/python/langgraph/local-server

Shows the expected output when the server starts, including the API URL, Studio UI URL, and API Docs URL.

```text
INFO:langgraph_api.cli:

        Welcome to

╦  ┌─┐┌┐┌┌─┐╔═╗┬─┐┌─┐┌─┐┬ ┬
║  ├─┤││││ ┬║ ╦├┬┘├─┤├─┘├─┤
╩═╝┴ ┴┘└┘└─┘╚═╝┴└─┴ ┴┴  ┴ ┴

- 🚀 API: http://127.0.0.1:2024
- 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
- 📚 API Docs: http://127.0.0.1:2024/docs

This in-memory server is designed for development and testing.
For production use, please use LangSmith Deployment.
```

--------------------------------

### Install HuggingFace chat model integration with pip

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the LangChain HuggingFace integration package using pip.

```bash
pip install -U "langchain[huggingface]"
```
Tokens:
Raw