tessl/pypi-langchain

Building applications with LLMs through composability

—

Pending

Overview

Eval results

Files

Persistence Patterns

Name: tessl/pypi-langchain
Author: tessl

Persistence enables agents to maintain state across multiple invocations, supporting conversation continuity, resumable execution, and cross-conversation data sharing.

Core Concepts

LangChain provides two mechanisms for persistence:

Checkpointers: Save and restore agent state within a conversation thread
Store: Share data across different conversation threads and agent executions

Checkpointers enable conversation continuity by persisting the message history and agent state. Store enables cross-thread data sharing for user preferences, caching, and global state.

Checkpointers

MemorySaver Checkpointer

The MemorySaver checkpointer stores conversation state in memory:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

Basic Usage:

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

# Create checkpointer
checkpointer = MemorySaver()

# Create agent with checkpointer
agent = create_agent(
    model="openai:gpt-4o",
    checkpointer=checkpointer
)

# First conversation turn
config = {"configurable": {"thread_id": "conversation-1"}}
result1 = agent.invoke({
    "messages": [HumanMessage(content="My name is Alice")]
}, config=config)

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

# Second conversation turn - agent remembers context
result2 = agent.invoke({
    "messages": [HumanMessage(content="What's my name?")]
}, config=config)

print(result2["messages"][-1].content)
# Output: "Your name is Alice."

Complete Example:

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langchain.tools import tool

@tool
def save_preference(key: str, value: str) -> str:
    """Save a user preference."""
    return f"Saved {key}={value}"

# Create agent with persistence
checkpointer = MemorySaver()
agent = create_agent(
    model="openai:gpt-4o",
    tools=[save_preference],
    checkpointer=checkpointer,
    system_prompt="You are a helpful assistant. Remember user preferences."
)

# Multi-turn conversation
config = {"configurable": {"thread_id": "user-123"}}

# Turn 1
result = agent.invoke({
    "messages": [HumanMessage(content="I prefer dark mode")]
}, config=config)
print(result["messages"][-1].content)

# Turn 2 - different session, same thread
result = agent.invoke({
    "messages": [HumanMessage(content="What are my preferences?")]
}, config=config)
print(result["messages"][-1].content)
# Agent remembers the preference from previous turn

Thread IDs for Conversation Continuity

Thread IDs uniquely identify conversation threads. Use them to maintain separate conversations:

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_agent(
    model="openai:gpt-4o",
    checkpointer=checkpointer
)

# Conversation with User A
config_a = {"configurable": {"thread_id": "user-a-thread"}}
agent.invoke({
    "messages": [HumanMessage(content="My favorite color is blue")]
}, config=config_a)

# Conversation with User B
config_b = {"configurable": {"thread_id": "user-b-thread"}}
agent.invoke({
    "messages": [HumanMessage(content="My favorite color is red")]
}, config=config_b)

# Continue User A's conversation
result_a = agent.invoke({
    "messages": [HumanMessage(content="What's my favorite color?")]
}, config=config_a)
print(result_a["messages"][-1].content)  # "Your favorite color is blue"

# Continue User B's conversation
result_b = agent.invoke({
    "messages": [HumanMessage(content="What's my favorite color?")]
}, config=config_b)
print(result_b["messages"][-1].content)  # "Your favorite color is red"

State Persistence

Checkpointers persist the entire agent state, including custom state fields:

from typing import TypedDict
from langchain.agents import create_agent, AgentState
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

# Define custom state
class CustomState(AgentState):
    user_name: str
    conversation_count: int

# Create agent with custom state
checkpointer = MemorySaver()
agent = create_agent(
    model="openai:gpt-4o",
    state_schema=CustomState,
    checkpointer=checkpointer
)

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

# First invocation with custom state
result = agent.invoke({
    "messages": [HumanMessage(content="Hello")],
    "user_name": "Alice",
    "conversation_count": 1
}, config=config)

# Second invocation - custom state is preserved
result = agent.invoke({
    "messages": [HumanMessage(content="How are you?")],
    "conversation_count": 2  # Update counter
}, config=config)

# Custom state persists across invocations
print(result.get("user_name"))  # "Alice" (preserved from first call)
print(result.get("conversation_count"))  # 2

Store for Cross-Thread Data

InMemoryStore

The InMemoryStore enables data sharing across conversation threads:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

Basic Store Operations:

from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import HumanMessage
from langgraph.store.memory import InMemoryStore
from langgraph.store import BaseStore
from langgraph.prebuilt import InjectedStore
from typing import Annotated

# Create store
store = InMemoryStore()

# Define tools that use store
@tool
def save_user_preference(
    key: str,
    value: str,
    store: Annotated[BaseStore, InjectedStore]
) -> str:
    """Save a user preference."""
    namespace = ("user_preferences",)
    store.put(namespace, key, {"value": value})
    return f"Saved {key}={value}"

@tool
def get_user_preference(
    key: str,
    store: Annotated[BaseStore, InjectedStore]
) -> str:
    """Get a user preference."""
    namespace = ("user_preferences",)
    item = store.get(namespace, key)
    if item:
        return f"{key}={item.value['value']}"
    return f"No preference found for {key}"

# Create agent with store
agent = create_agent(
    model="openai:gpt-4o",
    tools=[save_user_preference, get_user_preference],
    store=store
)

# Use in one thread
result1 = agent.invoke({
    "messages": [HumanMessage(content="Save my theme preference as dark")]
}, config={"configurable": {"thread_id": "thread-1"}})

# Access from different thread
result2 = agent.invoke({
    "messages": [HumanMessage(content="What is my theme preference?")]
}, config={"configurable": {"thread_id": "thread-2"}})

# Preference is shared across threads

Store Namespaces

Organize data using hierarchical namespaces:

from langchain.tools import tool
from langgraph.store import BaseStore
from langgraph.prebuilt import InjectedStore
from typing import Annotated

@tool
def save_setting(
    category: str,
    key: str,
    value: str,
    store: Annotated[BaseStore, InjectedStore]
) -> str:
    """Save a setting in a category."""
    # Use hierarchical namespace
    namespace = ("settings", category)
    store.put(namespace, key, {"value": value})
    return f"Saved {category}/{key}={value}"

@tool
def get_all_settings(
    category: str,
    store: Annotated[BaseStore, InjectedStore]
) -> str:
    """Get all settings in a category."""
    namespace = ("settings", category)
    items = store.search(namespace)

    if not items:
        return f"No settings in {category}"

    settings = []
    for item in items:
        key = item.key
        value = item.value["value"]
        settings.append(f"{key}={value}")

    return f"{category} settings: " + ", ".join(settings)

Caching with Store

Use store for caching expensive operations:

import hashlib
import json
import time
from langchain.tools import tool
from langgraph.store import BaseStore
from langgraph.prebuilt import InjectedStore
from typing import Annotated

@tool
def cached_api_call(
    endpoint: str,
    params: dict,
    store: Annotated[BaseStore, InjectedStore]
) -> dict:
    """Make an API call with caching."""
    # Create cache key
    cache_key = hashlib.md5(
        f"{endpoint}:{json.dumps(params, sort_keys=True)}".encode()
    ).hexdigest()

    namespace = ("api_cache",)

    # Check cache
    cached = store.get(namespace, cache_key)
    if cached:
        cache_age = time.time() - cached.value["timestamp"]
        if cache_age < 3600:  # 1 hour TTL
            return {
                "data": cached.value["data"],
                "cached": True,
                "age_seconds": cache_age
            }

    # Cache miss - make API call
    response = make_api_call(endpoint, params)

    # Store in cache
    store.put(namespace, cache_key, {
        "data": response,
        "timestamp": time.time()
    })

    return {
        "data": response,
        "cached": False
    }

def make_api_call(endpoint: str, params: dict) -> dict:
    """Simulate API call."""
    time.sleep(1)  # Simulate network delay
    return {"result": "data"}

Combining Checkpointer and Store

Use both checkpointer and store for comprehensive persistence:

from langchain.agents import create_agent, AgentState
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.store import BaseStore
from langgraph.prebuilt import InjectedStore, InjectedState
from typing import Annotated, TypedDict

# Custom state
class UserState(AgentState):
    user_id: str

# Tools using both state and store
@tool
def save_note(
    note: str,
    state: Annotated[UserState, InjectedState],
    store: Annotated[BaseStore, InjectedStore]
) -> str:
    """Save a note for the current user."""
    user_id = state.get("user_id", "unknown")
    namespace = ("user_notes", user_id)

    # Generate note ID
    import time
    note_id = f"note_{int(time.time())}"

    store.put(namespace, note_id, {
        "note": note,
        "timestamp": time.time()
    })

    return f"Note saved: {note}"

@tool
def list_notes(
    state: Annotated[UserState, InjectedState],
    store: Annotated[BaseStore, InjectedStore]
) -> str:
    """List all notes for the current user."""
    user_id = state.get("user_id", "unknown")
    namespace = ("user_notes", user_id)

    items = store.search(namespace)
    if not items:
        return "No notes found"

    notes = []
    for item in items:
        note_data = item.value
        notes.append(f"- {note_data['note']}")

    return "Your notes:\n" + "\n".join(notes)

# Create agent with both persistence mechanisms
checkpointer = MemorySaver()
store = InMemoryStore()

agent = create_agent(
    model="openai:gpt-4o",
    tools=[save_note, list_notes],
    state_schema=UserState,
    checkpointer=checkpointer,  # Thread-specific state
    store=store  # Cross-thread data
)

# User 1's conversation
config_user1 = {"configurable": {"thread_id": "user1-thread"}}
agent.invoke({
    "messages": [HumanMessage(content="Save a note: Buy groceries")],
    "user_id": "user-1"
}, config=config_user1)

# User 2's conversation (different thread)
config_user2 = {"configurable": {"thread_id": "user2-thread"}}
agent.invoke({
    "messages": [HumanMessage(content="Save a note: Call mom")],
    "user_id": "user-2"
}, config=config_user2)

# Each user has their own notes (stored cross-thread)
# Each user has their own conversation history (checkpointed)

Persistence with Streaming

Combine persistence with streaming for responsive, stateful interactions:

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_agent(
    model="openai:gpt-4o",
    checkpointer=checkpointer
)

config = {"configurable": {"thread_id": "stream-thread"}}

# Stream first turn
print("Turn 1:")
for chunk in agent.stream({
    "messages": [HumanMessage(content="My favorite animal is a cat")]
}, config=config):
    if "agent" in chunk:
        messages = chunk["agent"].get("messages", [])
        for msg in messages:
            if hasattr(msg, 'content') and msg.content:
                print(msg.content, end="", flush=True)

# Stream second turn - state persisted
print("\n\nTurn 2:")
for chunk in agent.stream({
    "messages": [HumanMessage(content="What's my favorite animal?")]
}, config=config):
    if "agent" in chunk:
        messages = chunk["agent"].get("messages", [])
        for msg in messages:
            if hasattr(msg, 'content') and msg.content:
                print(msg.content, end="", flush=True)

Best Practices

Use Unique Thread IDs

Generate unique, consistent thread IDs for each conversation:

import uuid

# Per-user threads
thread_id = f"user-{user_id}-{session_id}"

# Or use UUIDs
thread_id = str(uuid.uuid4())

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

Clean Up Old State

Periodically clean up old checkpoints and store data:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

# Checkpointer automatically manages memory
# For production, use database-backed checkpointers with TTL

Namespace Store Data Clearly

Use descriptive, hierarchical namespaces:

# Good namespaces
namespace = ("user_data", user_id, "preferences")
namespace = ("cache", "api_responses")
namespace = ("analytics", "daily_stats", date_str)

# Avoid flat namespaces
namespace = ("data",)  # Too generic

Common Mistakes

Mistake: Forgetting Thread ID

# Wrong - no thread ID means no persistence
result = agent.invoke({
    "messages": [HumanMessage(content="Remember this")]
})

# Right - include thread ID
config = {"configurable": {"thread_id": "user-123"}}
result = agent.invoke({
    "messages": [HumanMessage(content="Remember this")]
}, config=config)

Mistake: Mixing Thread Data

# Wrong - reusing thread ID for different users
config = {"configurable": {"thread_id": "conversation"}}

# Right - unique thread ID per user
config_user1 = {"configurable": {"thread_id": f"user-{user1_id}"}}
config_user2 = {"configurable": {"thread_id": f"user-{user2_id}"}}

Mistake: Not Using Store for Cross-Thread Data

# Wrong - trying to share data via checkpointer
# Checkpointer is thread-specific only

# Right - use store for cross-thread data
store = InMemoryStore()
agent = create_agent(
    model="openai:gpt-4o",
    checkpointer=MemorySaver(),  # Thread-specific
    store=store  # Cross-thread
)