How to handle multiple agents with Checkpoints?

[samidarko]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Annotated

from langchain_core.messages import SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from langgraph.graph.graph import CompiledGraph
from langgraph.graph.message import add_messages
from langgraph.types import Checkpointer
from langgraph.utils.runnable import RunnableCallable
from typing_extensions import TypedDict



class State(TypedDict):
    messages: Annotated[list, add_messages]

def create_basic_agent(
    system_prompt: str, checkpointer: Checkpointer = None
) -> CompiledGraph:
    prompt = ChatPromptTemplate.from_messages(
        [
            SystemMessage(content=system_prompt),
            MessagesPlaceholder(variable_name="messages"),
        ]
    )

    agent = prompt | llm

    def call_model(state: State):
        return {"messages": [agent.invoke(state)]}

    async def acall_model(state: State):
        return {"messages": [await agent.ainvoke(state)]}

    graph_builder = StateGraph(State)
    graph_builder.add_node("agent", RunnableCallable(call_model, acall_model))
    graph_builder.set_entry_point("agent")
    graph_builder.set_finish_point("agent")

    graph = graph_builder.compile(checkpointer=checkpointer)

    return graph

def main():
    with PostgresSaver.from_conn_string(
        "postgresql://postgres:[email protected]:5432/checkpoint"
    ) as checkpointer:
        checkpointer.setup()

        agent = create_basic_agent(system_prompt="you are an helpful assistant", checkpointer=checkpointer)
        config = {"configurable": {"thread_id": "3fa85f64-5717-4562-b3fc-2c963f66afa8", "checkpoint_ns": "agent"}}
        result = agent.invoke(
            input={"messages": {"type": "user", "content": "hi!"}},
            config=config
        )
        print(result)


if __name__ == '__main__':
    main()

Description

I have multiple agents and I'm not sure if I should have multiple checkpoint tables, one for each of them, or only one table. The latter seems more reasonable, even more if I had another table for Conversations (a.k.a. treads) but I'm not even sure I need one as well?

I was thinking of using checkpoint_ns to differentiate them but it doesn't seem to work.

In the example I'm providing I'm calling invoke() with a config that contains checkpoint_ns key with agent for value but the column is empty.

checkpoint=# select thread_id, checkpoint_ns from checkpoints;
              thread_id               | checkpoint_ns
--------------------------------------+---------------
 3fa85f64-5717-4562-b3fc-2c963f66afa8 |

What am I doing wrong? Any guidance would be appreciated.

System Info

langchain==0.3.9
langchain-core==0.3.21
langchain-openai==0.2.10
langchain-text-splitters==0.3.2
langgraph==0.2.53
langgraph-checkpoint==2.0.7
langgraph-checkpoint-postgres==2.0.6
langgraph-sdk==0.1.40
langserve==0.3.0
langsmith==0.1.147

[samidarko]

I'm closing this as I moved it to the langgraph repository langchain-ai/langgraph#2676