Langgraph checkpointer selective memory

Product Help LangGraph
1

I am trying to add a selective memory for my state which have fields like query, answer, context_documentsand other internal fields. I want to selective persist only query and answer for checkpointing. Is there a documentation or way to do that or I would have to extract query, answer in a single messages list with roles with add operator as shown in documentation? I don’t want to persist long context documents in the checkpointing.

2

Hi @ritik

AFAK, there isn’t a built‑in toggle to exclude specific state keys from checkpointing. A checkpointer persists the state values you define. To avoid saving large context_documents , don’t put them in the State you persist; keep only minimal fields (e.g., query , answer ) in state, and handle large data either inside a node without returning it, or store it externally and reference it by ID.

Minimal state

class State(TypedDict):
    query: str
    answer: str

def retrieve_and_answer(state: State) -> dict:
    # Retrieve docs locally inside the node
    docs = retriever.invoke(state["query"])  # not returned → not persisted
    answer = llm.invoke(format_prompt(state["query"], docs))
    return {"answer": answer}

Store large data in a Store, persist only IDs

class State(TypedDict):
    query: str
    context_id: str  # small pointer only
    answer: str

def fetch_context(state: State, *, store: BaseStore, config: RunnableConfig) -> dict:
    docs = retriever.invoke(state["query"])  # large
    ctx_id = str(uuid4())
    ns = (config["configurable"]["thread_id"], "ctx")
    store.put(ns, ctx_id, {"docs": docs})  # heavy data offloaded
    return {"context_id": ctx_id}

def answer_with_context(state: State, *, store: BaseStore, config: RunnableConfig) -> dict:
    ns = (config["configurable"]["thread_id"], "ctx")
    # Retrieve by id (via store.search or equivalent get)
    docs = store.search(ns, query=state["context_id"])  # implement lookup pattern you prefer
    answer = llm.invoke(format_prompt(state["query"], docs))
    return {"answer": answer}

Filter at persistence time (advanced)

# Sketch: wrap a saver to drop unwanted keys before persisting
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.base import BaseCheckpointSaver

class FilteringSaver(BaseCheckpointSaver):
    def __init__(self, inner: BaseCheckpointSaver, keep_keys=("query", "answer")):
        self.inner = inner
        self.keep = set(keep_keys)

    # Delegate required methods; in the write path, filter values
    def put(self, *args, **kwargs):
        # Depending on the exact saver signature, access the checkpoint's values
        # and keep only keys in self.keep before delegating to self.inner.put(...)
        return self.inner.put(*args, **kwargs)  # after filtering

checkpointer = FilteringSaver(SqliteSaver.from_conn_string("sqlite:///checkpoints.db"))
graph = builder.compile(checkpointer=checkpointer)

References

1 Like
3

Pawel, so running langgraph localhost but the added dependency for langgraph.checkpoint.sqlite .. will use that for long term memory store .. postgre more for production (presume when deployed to langgraph platform in production that is more suited)

1 Like