WEBDIENSTE/langgraph

Fork 0

mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-08 02:37:52 +02:00

Files

T

Simon FrankandGitHub 85c809a651 docs: fixed a wrong import in persistence docs (#5045 )

2025-06-10 20:53:50 +00:00

73 KiB

Raw Blame History

Add persistence

Many AI applications need memory to share context across multiple interactions. LangGraph supports two types of memory essential for building conversational agents:

Short-term memory: Tracks the ongoing conversation by maintaining message history within a session.
Long-term memory: Stores user-specific or application-level data across sessions.

Terminology

In LangGraph:

Short-term memory is also referred to as thread-level memory.

Long-term memory is also called cross-thread memory.

A thread represents a sequence of related runs grouped by the same thread_id.

In [1]:

# hide-cell
%pip install --quiet -U langgraph "langchain[anthropic]"

In [2]:

# hide-cell
import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

ANTHROPIC_API_KEY:  ········

Add short-term memory

Short-term memory (thread-level persistence) enables agents to track multi-turn conversations. To add short-term memory:

In [18]:

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START

# highlight-next-line
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")


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")

checkpointer = InMemorySaver()
# highlight-next-line
graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        # highlight-next-line
        "thread_id": "1"
    }
}

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    # highlight-next-line
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    # highlight-next-line
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

================================[1m Human Message [0m=================================

hi! I'm bob
==================================[1m Ai Message [0m==================================

Hi Bob! How are you doing today? Is there anything I can help you with?
================================[1m Human Message [0m=================================

what's my name?
==================================[1m Ai Message [0m==================================

Your name is Bob.

!!! info "Not needed for LangGraph API users"

If you're using the LangGraph API, **don't need** to provide checkpointer when compiling the graph. The API automatically handles checkpointing for you.

Use in production

In production, you would want to use a checkpointer backed by a database:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
# highlight-next-line
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    builder = StateGraph(...)
    # highlight-next-line
    graph = builder.compile(checkpointer=checkpointer)

??? example "Example: using Postgres checkpointer"

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

!!! Setup
    You need to call `checkpointer.setup()` the first time you're using Postgres checkpointer

=== "Sync"

    ```python
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    # highlight-next-line
    from langgraph.checkpoint.postgres import PostgresSaver
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
    # highlight-next-line
    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")
        
        # highlight-next-line
        graph = builder.compile(checkpointer=checkpointer)
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1"
            }
        }
    
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
        
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "what's my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
    ```

=== "Async"

    ```python
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    # highlight-next-line
    from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
    # highlight-next-line
    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")
        
        # highlight-next-line
        graph = builder.compile(checkpointer=checkpointer)
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1"
            }
        }
    
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
        
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "what's my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
    ```

??? example "Example: using MongoDB checkpointer"

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

!!! note "Setup"

    To use the MongoDB checkpointer, you will need a MongoDB cluster. Follow [this guide](https://www.mongodb.com/docs/guides/atlas/cluster/) to create a cluster if you don't already have one.

=== "Sync"

    ```python
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    # highlight-next-line
    from langgraph.checkpoint.mongodb import MongoDBSaver
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "localhost:27017"
    # highlight-next-line
    with MongoDBSaver.from_conn_string(DB_URI) as checkpointer:
    
        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")
        
        # highlight-next-line
        graph = builder.compile(checkpointer=checkpointer)
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1"
            }
        }
    
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
        
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "what's my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
    ```

=== "Async"

    ```python
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    # highlight-next-line
    from langgraph.checkpoint.mongodb.aio import AsyncMongoDBSaver
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "localhost:27017"
    # highlight-next-line
    async with AsyncMongoDBSaver.from_conn_string(DB_URI) as checkpointer:
    
        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")
        
        # highlight-next-line
        graph = builder.compile(checkpointer=checkpointer)
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1"
            }
        }
    
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
        
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "what's my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
    ```

??? example "Example: using Redis checkpointer"

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

!!! Setup
    You need to call `checkpointer.setup()` the first time you're using Redis checkpointer


=== "Sync"

    ```python
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    # highlight-next-line
    from langgraph.checkpoint.redis import RedisSaver
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "redis://localhost:6379"
    # highlight-next-line
    with RedisSaver.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")
        
        # highlight-next-line
        graph = builder.compile(checkpointer=checkpointer)
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1"
            }
        }
    
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
        
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "what's my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
    ```

=== "Async"

    ```python
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    # highlight-next-line
    from langgraph.checkpoint.redis.aio import AsyncRedisSaver
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "redis://localhost:6379"
    # highlight-next-line
    async with AsyncRedisSaver.from_conn_string(DB_URI) as checkpointer:
        # await checkpointer.asetup()
    
        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")
        
        # highlight-next-line
        graph = builder.compile(checkpointer=checkpointer)
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1"
            }
        }
    
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()
        
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "what's my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values"
        ):
            chunk["messages"][-1].pretty_print()     
    ```

Use with subgraphs

If your graph contains subgraphs, you only need to provide the checkpointer when compiling the parent graph. LangGraph will automatically propagate the checkpointer to the child subgraphs.

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

class State(TypedDict):
    foo: str

# Subgraph

def subgraph_node_1(state: State):
    return {"foo": state["foo"] + "bar"}

subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
# highlight-next-line
subgraph = subgraph_builder.compile()

# Parent graph

def node_1(state: State):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(State)
# highlight-next-line
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")

checkpointer = InMemorySaver()
# highlight-next-line
graph = builder.compile(checkpointer=checkpointer)

If you want the subgraph to have its own memory, you can compile it with checkpointer=True. This is useful in multi-agent systems, if you want agents to keep track of their internal message histories:

subgraph_builder = StateGraph(...)
# highlight-next-line
subgraph = subgraph_builder.compile(checkpointer=True)

Use with Functional API

To add short-term memory to a Functional API LangGraph workflow:

Pass checkpointer instance to the [entrypoint()][langgraph.func.entrypoint] decorator:

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def workflow(inputs)
    ...

Optionally expose previous parameter in the workflow function signature:

@entrypoint(checkpointer=checkpointer)
def workflow(
    inputs,
    *,
    # you can optionally specify `previous` in the workflow function signature
    # to access the return value from the workflow as of the last execution
    previous
):
    previous = previous or []
    combined_inputs = previous + inputs
    result = do_something(combined_inputs)
    ...

Optionally choose which values will be returned from the workflow and which will be saved by the checkpointer as previous:

@entrypoint(checkpointer=checkpointer)
def workflow(inputs, *, previous):
    ...
    result = do_something(...)
    return entrypoint.final(value=result, save=combine(inputs, result))

??? example "Example: add short-term memory to Functional API workflow"

```python
from langchain_core.messages import AnyMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

# highlight-next-line
@task
def call_model(messages: list[AnyMessage]):
    response = model.invoke(messages)
    return response

checkpointer = InMemorySaver()

# highlight-next-line
@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[AnyMessage], *, previous: list[AnyMessage]):
    if previous:
        inputs = add_messages(previous, inputs)

    response = call_model(inputs).result()
    return entrypoint.final(value=response, save=add_messages(inputs, response))

config = {
    "configurable": {
        # highlight-next-line
        "thread_id": "1"
    }
}

for chunk in workflow.invoke(
    [{"role": "user", "content": "hi! I'm bob"}],
    # highlight-next-line
    config,
    stream_mode="values",
):
    chunk.pretty_print()

for chunk in workflow.stream(
    [{"role": "user", "content": "what's my name?"}],
    # highlight-next-line
    config,
    stream_mode="values",
):
    chunk.pretty_print()
```

Manage checkpoints

You can view and delete the information stored by the checkpointer:

??? "View thread state (checkpoint)"

=== "Graph/Functional API"

    ```python
    config = {
        "configurable": {
            # highlight-next-line
            "thread_id": "1",
            # optionally provide an ID for a specific checkpoint,
            # otherwise the latest checkpoint is shown
            # highlight-next-line
            # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"
              
        }
    }
    # highlight-next-line
    graph.get_state(config)
    ```

    ```
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, next=(), 
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
        metadata={
            'source': 'loop',
            'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
            'step': 4,
            'parents': {},
            'thread_id': '1'
        },
        created_at='2025-05-05T16:01:24.680462+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}}, 
        tasks=(),
        interrupts=()
    )
    ```

=== "Checkpointer API"

    ```python
    config = {
        "configurable": {
            # highlight-next-line
            "thread_id": "1",
            # optionally provide an ID for a specific checkpoint,
            # otherwise the latest checkpoint is shown
            # highlight-next-line
            # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"
              
        }
    }
    # highlight-next-line
    checkpointer.get_tuple(config)
    ```

    ```
    CheckpointTuple(
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
        checkpoint={
            'v': 3,
            'ts': '2025-05-05T16:01:24.680462+00:00',
            'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a',
            'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},
            'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},
        },
        metadata={
            'source': 'loop',
            'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
            'step': 4,
            'parents': {},
            'thread_id': '1'
        },
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        pending_writes=[]
    )
    ```

??? "View the history of the thread (checkpoints)"

=== "Graph/Functional API"

    ```python
    config = {
        "configurable": {
            # highlight-next-line
            "thread_id": "1"
        }
    }
    # highlight-next-line
    list(graph.get_state_history(config))
    ```

    ```
    [
        StateSnapshot(
            values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, 
            next=(), 
            config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}}, 
            metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'},
            created_at='2025-05-05T16:01:24.680462+00:00',
            parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
            tasks=(),
            interrupts=()
        ),
        StateSnapshot(
            values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")]}, 
            next=('call_model',), 
            config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
            metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'},
            created_at='2025-05-05T16:01:23.863421+00:00',
            parent_config={...}
            tasks=(PregelTask(id='8ab4155e-6b15-b885-9ce5-bed69a2c305c', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Your name is Bob.')}),),
            interrupts=()
        ),
        StateSnapshot(
            values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, 
            next=('__start__',), 
            config={...}, 
            metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'},
            created_at='2025-05-05T16:01:23.863173+00:00',
            parent_config={...}
            tasks=(PregelTask(id='24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "what's my name?"}]}),),
            interrupts=()
        ),
        StateSnapshot(
            values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, 
            next=(), 
            config={...}, 
            metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'},
            created_at='2025-05-05T16:01:23.862295+00:00',
            parent_config={...}
            tasks=(),
            interrupts=()
        ),
        StateSnapshot(
            values={'messages': [HumanMessage(content="hi! I'm bob")]}, 
            next=('call_model',), 
            config={...}, 
            metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'}, 
            created_at='2025-05-05T16:01:22.278960+00:00', 
            parent_config={...}
            tasks=(PregelTask(id='8cbd75e0-3720-b056-04f7-71ac805140a0', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}),), 
            interrupts=()
        ),
        StateSnapshot(
            values={'messages': []}, 
            next=('__start__',), 
            config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}},
            metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'}, 
            created_at='2025-05-05T16:01:22.277497+00:00', 
            parent_config=None,
            tasks=(PregelTask(id='d458367b-8265-812c-18e2-33001d199ce6', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}),), 
            interrupts=()
        )
    ]       
    ```

=== "Checkpointer API"

    ```python
    config = {
        "configurable": {
            # highlight-next-line
            "thread_id": "1"
        }
    }
    # highlight-next-line
    list(checkpointer.list(config))
    ```

    ```
    [
        CheckpointTuple(
            config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}}, 
            checkpoint={
                'v': 3, 
                'ts': '2025-05-05T16:01:24.680462+00:00', 
                'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a', 
                'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, 
                'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},
                'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},
            },
            metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'}, 
            parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}}, 
            pending_writes=[]
        ),
        CheckpointTuple(
            config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
            checkpoint={
                'v': 3, 
                'ts': '2025-05-05T16:01:23.863421+00:00', 
                'id': '1f029ca3-1790-6b0a-8003-baf965b6a38f', 
                'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000005.0.7935064215293443', 'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}, 
                'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, 
                'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")], 'branch:to:call_model': None}
            }, 
            metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'}, 
            parent_config={...}, 
            pending_writes=[('8ab4155e-6b15-b885-9ce5-bed69a2c305c', 'messages', AIMessage(content='Your name is Bob.'))]
        ),
        CheckpointTuple(
            config={...}, 
            checkpoint={
                'v': 3, 
                'ts': '2025-05-05T16:01:23.863173+00:00', 
                'id': '1f029ca3-1790-616e-8002-9e021694a0cd', 
                'channel_versions': {'__start__': '00000000000000000000000000000004.0.5736472536395331', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, 
                'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, 
                'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}, 'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}
            }, 
            metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'}, 
            parent_config={...}, 
            pending_writes=[('24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', 'messages', [{'role': 'user', 'content': "what's my name?"}]), ('24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', 'branch:to:call_model', None)]
        ),
        CheckpointTuple(
            config={...}, 
            checkpoint={
                'v': 3, 
                'ts': '2025-05-05T16:01:23.862295+00:00', 
                'id': '1f029ca3-178d-6f54-8001-d7b180db0c89', 
                'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, 
                'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, 
                'channel_values': {'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}
            }, 
            metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'}, 
            parent_config={...}, 
            pending_writes=[]
        ),
        CheckpointTuple(
            config={...}, 
            checkpoint={
                'v': 3, 
                'ts': '2025-05-05T16:01:22.278960+00:00', 
                'id': '1f029ca3-0874-6612-8000-339f2abc83b1', 
                'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000002.0.30296526818059655', 'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}, 
                'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}}, 
                'channel_values': {'messages': [HumanMessage(content="hi! I'm bob")], 'branch:to:call_model': None}
            }, 
            metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'}, 
            parent_config={...}, 
            pending_writes=[('8cbd75e0-3720-b056-04f7-71ac805140a0', 'messages', AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'))]
        ),
        CheckpointTuple(
            config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}}, 
            checkpoint={
                'v': 3, 
                'ts': '2025-05-05T16:01:22.277497+00:00', 
                'id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565', 
                'channel_versions': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 
                'versions_seen': {'__input__': {}}, 
                'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}
            }, 
            metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'}, 
            parent_config=None, 
            pending_writes=[('d458367b-8265-812c-18e2-33001d199ce6', 'messages', [{'role': 'user', 'content': "hi! I'm bob"}]), ('d458367b-8265-812c-18e2-33001d199ce6', 'branch:to:call_model', None)]
        )
    ]
    ```

??? "Delete all checkpoints for a thread"

```python
thread_id = "1"
checkpointer.delete_thread(thread_id)
```

Add long-term memory

Use long-term memory (cross-thread persistence) to store user-specific or application-specific data across conversations. This is useful for applications like chatbots, where you want to remember user preferences or other information.

To use long-term memory, we need to [provide a store][langgraph.store.base.BaseStore] when creating the graph:

In [5]:

import uuid
from typing_extensions import Annotated, TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver

# highlight-next-line
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore

model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")


def call_model(
    state: MessagesState,
    config: RunnableConfig,
    *,
    # highlight-next-line
    store: BaseStore,  # (1)!
):
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)
    # highlight-next-line
    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"
        # highlight-next-line
        store.put(namespace, str(uuid.uuid4()), {"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")

checkpointer = InMemorySaver()
store = InMemoryStore()

graph = builder.compile(
    checkpointer=checkpointer,
    # highlight-next-line
    store=store,
)

This is the store we compiled the graph with

In [6]:

config = {
    "configurable": {
        # highlight-next-line
        "thread_id": "1",
        # highlight-next-line
        "user_id": "1",
    }
}
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
    # highlight-next-line
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

config = {
    "configurable": {
        # highlight-next-line
        "thread_id": "2",
        "user_id": "1",
    }
}

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "what is my name?"}]},
    # highlight-next-line
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

================================[1m Human Message [0m=================================

Hi! Remember: my name is Bob
==================================[1m Ai Message [0m==================================

Hi Bob! I'll remember that your name is Bob. How are you doing today?
================================[1m Human Message [0m=================================

what is my name?
==================================[1m Ai Message [0m==================================

Your name is Bob.

!!! info "Not needed for LangGraph API users"

If you're using the LangGraph API, **don't need** to provide store when compiling the graph. The API automatically handles storage infrastructure for you.

Use in production

In production, you would want to use a store backed by a database:

from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
# highlight-next-line
with PostgresStore.from_conn_string(DB_URI) as store:
    builder = StateGraph(...)
    # highlight-next-line
    graph = builder.compile(store=store)

??? example "Example: using Postgres store"

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

!!! Setup
    You need to call `store.setup()` the first time you're using Postgres store

=== "Sync"

    ```python
    from langchain_core.runnables import RunnableConfig
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    from langgraph.checkpoint.postgres import PostgresSaver
    # highlight-next-line
    from langgraph.store.postgres import PostgresStore
    from langgraph.store.base import BaseStore
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
    
    with (
        # highlight-next-line
        PostgresStore.from_conn_string(DB_URI) as store,
        PostgresSaver.from_conn_string(DB_URI) as checkpointer,
    ):
        # store.setup()
        # checkpointer.setup()
    
        def call_model(
            state: MessagesState,
            config: RunnableConfig,
            *,
            # highlight-next-line
            store: BaseStore,
        ):
            user_id = config["configurable"]["user_id"]
            namespace = ("memories", user_id)
            # highlight-next-line
            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"
                # highlight-next-line
                store.put(namespace, str(uuid.uuid4()), {"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,
            # highlight-next-line
            store=store,
        )
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1",
                # highlight-next-line
                "user_id": "1",
            }
        }
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
        
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "2",
                "user_id": "1",
            }
        }
    
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "what is my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
    ```

=== "Async"

    ```python
    from langchain_core.runnables import RunnableConfig
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
    # highlight-next-line
    from langgraph.store.postgres.aio import AsyncPostgresStore
    from langgraph.store.base import BaseStore
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
    
    async with (
        # highlight-next-line
        AsyncPostgresStore.from_conn_string(DB_URI) as store,
        AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer,
    ):
        # await store.setup()
        # await checkpointer.setup()
    
        async def call_model(
            state: MessagesState,
            config: RunnableConfig,
            *,
            # highlight-next-line
            store: BaseStore,
        ):
            user_id = config["configurable"]["user_id"]
            namespace = ("memories", user_id)
            # highlight-next-line
            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"
                # highlight-next-line
                await store.aput(namespace, str(uuid.uuid4()), {"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,
            # highlight-next-line
            store=store,
        )
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1",
                # highlight-next-line
                "user_id": "1",
            }
        }
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
        
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "2",
                "user_id": "1",
            }
        }
    
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "what is my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
    ```

??? example "Example: using Redis store"

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

!!! Setup
    You need to call `store.setup()` the first time you're using Redis store


=== "Sync"

    ```python
    from langchain_core.runnables import RunnableConfig
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    from langgraph.checkpoint.redis import RedisSaver
    # highlight-next-line
    from langgraph.store.redis import RedisStore
    from langgraph.store.base import BaseStore
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
        
    DB_URI = "redis://localhost:6379"
    
    with (
        # highlight-next-line
        RedisStore.from_conn_string(DB_URI) as store,
        RedisSaver.from_conn_string(DB_URI) as checkpointer,
    ):
        store.setup()
        checkpointer.setup()
    
        def call_model(
            state: MessagesState,
            config: RunnableConfig,
            *,
            # highlight-next-line
            store: BaseStore,
        ):
            user_id = config["configurable"]["user_id"]
            namespace = ("memories", user_id)
            # highlight-next-line
            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"
                # highlight-next-line
                store.put(namespace, str(uuid.uuid4()), {"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,
            # highlight-next-line
            store=store,
        )
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1",
                # highlight-next-line
                "user_id": "1",
            }
        }
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
        
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "2",
                "user_id": "1",
            }
        }
    
        for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "what is my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
    ```

=== "Async"

    ```python
    from langchain_core.runnables import RunnableConfig
    from langchain.chat_models import init_chat_model
    from langgraph.graph import StateGraph, MessagesState, START
    from langgraph.checkpoint.redis.aio import AsyncRedisSaver
    # highlight-next-line
    from langgraph.store.redis.aio import AsyncRedisStore
    from langgraph.store.base import BaseStore
    
    model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
    
    DB_URI = "redis://localhost:6379"
    
    async with (
        # highlight-next-line
        AsyncRedisStore.from_conn_string(DB_URI) as store,
        AsyncRedisSaver.from_conn_string(DB_URI) as checkpointer,
    ):
        # await store.setup()
        # await checkpointer.asetup()
    
        async def call_model(
            state: MessagesState,
            config: RunnableConfig,
            *,
            # highlight-next-line
            store: BaseStore,
        ):
            user_id = config["configurable"]["user_id"]
            namespace = ("memories", user_id)
            # highlight-next-line
            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"
                # highlight-next-line
                await store.aput(namespace, str(uuid.uuid4()), {"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,
            # highlight-next-line
            store=store,
        )
    
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "1",
                # highlight-next-line
                "user_id": "1",
            }
        }
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()
        
        config = {
            "configurable": {
                # highlight-next-line
                "thread_id": "2",
                "user_id": "1",
            }
        }
    
        async for chunk in graph.astream(
            {"messages": [{"role": "user", "content": "what is my name?"}]},
            # highlight-next-line
            config,
            stream_mode="values",
        ):
            chunk["messages"][-1].pretty_print()  
    ```

Use semantic search

You can enable semantic search in your graph's memory store: this lets graph agent search for items in the store by semantic similarity.

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

# Create store with semantic search enabled
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 1536,
    }
)

store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})

items = store.search(
    ("user_123", "memories"), query="I'm hungry", limit=1
)

??? example "Long-term memory with semantic search"

```python
from typing import Optional

from langchain.embeddings import init_embeddings
from langchain.chat_models import init_chat_model
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.graph import START, MessagesState, StateGraph

llm = init_chat_model("openai:gpt-4o-mini")

# Create store with semantic search enabled
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 1536,
    }
)

store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})

def chat(state, *, store: BaseStore):
    # Search based on user's last message
    items = store.search(
        ("user_123", "memories"), query=state["messages"][-1].content, limit=2
    )
    memories = "\n".join(item.value["text"] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    response = llm.invoke(
        [
            {"role": "system", "content": f"You are a helpful assistant.\n{memories}"},
            *state["messages"],
        ]
    )
    return {"messages": [response]}


builder = StateGraph(MessagesState)
builder.add_node(chat)
builder.add_edge(START, "chat")
graph = builder.compile(store=store)

for message, metadata in graph.stream(
    input={"messages": [{"role": "user", "content": "I'm hungry"}]},
    stream_mode="messages",
):
    print(message.content, end="")
```

See this guide for more information on how to use semantic search with LangGraph memory store.

73 KiB Raw Blame History

Add persistence

Add short-term memory

Use in production

Use with subgraphs

Use with Functional API

Manage checkpoints

Add long-term memory

Use in production

Use semantic search

73 KiB

Raw Blame History